mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-28 00:53:45 +08:00
Restore GM Cancel Remapping
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
"""Independent lateral axes for exact source-qualified Bolt and Volt owners."""
|
||||
|
||||
import math
|
||||
from opendbc.car.gm.cancel_remap import aol_enabled, valid_alternative_experience
|
||||
|
||||
from opendbc.car.structs import CarParams
|
||||
from opendbc.car.gm.values import (malibu_hybrid_profile, volt_cc_pedal_profile, control_flags,
|
||||
@@ -44,7 +45,7 @@ def qualified_gm(cp) -> bool:
|
||||
cp.steerControlType != CarParams.SteerControlType.torque or
|
||||
len(cp.safetyConfigs) != 1 or cp.safetyConfigs[0].safetyModel != CarParams.SafetyModel.gm or
|
||||
int(cp.safetyConfigs[0].safetyParam) not in GM_AOL_WORDS or
|
||||
int(cp.alternativeExperience) not in (0, GM_AOL_ALTERNATIVE_EXPERIENCE) or
|
||||
not valid_alternative_experience(cp) or
|
||||
cp.lateralTuning.which() not in ('pid', 'torque')):
|
||||
return False
|
||||
if cp.lateralTuning.which() == 'torque':
|
||||
@@ -95,7 +96,7 @@ def qualified_gm(cp) -> bool:
|
||||
|
||||
def lateral_request(cp, cc) -> bool:
|
||||
"""Only a finalized independent-axis owner may relax its cruise-active lateral gate."""
|
||||
return (qualified_gm(cp) and int(cp.alternativeExperience) == GM_AOL_ALTERNATIVE_EXPERIENCE and cc.latActive)
|
||||
return (qualified_gm(cp) and aol_enabled(cp) and cc.latActive)
|
||||
|
||||
|
||||
def native_bootstrap_supported(cp) -> bool:
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
"""Exact optional Bolt Cancel secondary-input configuration, selected at startup."""
|
||||
from opendbc.car.gm.values import BOLT_PEDAL_WORDS, is_bolt_pedal_profile
|
||||
|
||||
GM_REMAP_CANCEL_TO_DISTANCE = 64
|
||||
KEY = "RemapCancelToDistance"
|
||||
|
||||
|
||||
def supported(cp):
|
||||
try:
|
||||
return bool(cp.carFingerprint in BOLT_PEDAL_WORDS and is_bolt_pedal_profile(cp))
|
||||
except (AttributeError, IndexError, TypeError, ValueError):
|
||||
return False
|
||||
|
||||
|
||||
def valid_alternative_experience(cp):
|
||||
try:
|
||||
value = int(cp.alternativeExperience)
|
||||
return value in (0, 32) or value in (64, 96) and supported(cp)
|
||||
except (AttributeError, TypeError, ValueError):
|
||||
return False
|
||||
|
||||
|
||||
def enabled(cp):
|
||||
try:
|
||||
return bool(supported(cp) and int(cp.alternativeExperience) in (64, 96))
|
||||
except (AttributeError, TypeError, ValueError):
|
||||
return False
|
||||
|
||||
|
||||
def aol_enabled(cp):
|
||||
return valid_alternative_experience(cp) and int(cp.alternativeExperience) in (32, 96)
|
||||
|
||||
|
||||
def prepare(cp, selected):
|
||||
if cp.brand != 'gm' or cp.carFingerprint not in BOLT_PEDAL_WORDS:
|
||||
return
|
||||
cp.alternativeExperience &= ~GM_REMAP_CANCEL_TO_DISTANCE
|
||||
if selected and supported(cp):
|
||||
cp.alternativeExperience |= GM_REMAP_CANCEL_TO_DISTANCE
|
||||
|
||||
|
||||
def migrate_assignments(params, cp):
|
||||
from openpilot.starpilot.saved_source import read_saved
|
||||
if not enabled(cp):
|
||||
return False
|
||||
marker, readable = read_saved(params, 'CancelButtonControlsMigrated', 8)
|
||||
if not readable or marker not in (None, b'0'):
|
||||
return False
|
||||
pairs = (('DistanceButtonControl', 'CancelButtonControl', 1),
|
||||
('LongDistanceButtonControl', 'LongCancelButtonControl', 5),
|
||||
('VeryLongDistanceButtonControl', 'VeryLongCancelButtonControl', 6))
|
||||
pending = []
|
||||
valid = tuple(str(value).encode() for value in range(15))
|
||||
for source, target, default in pairs:
|
||||
raw, readable = read_saved(params, target, 8)
|
||||
previous, source_readable = read_saved(params, source, 8)
|
||||
if not readable or raw is not None and raw not in valid or not source_readable or previous is not None and previous not in valid:
|
||||
return False
|
||||
if raw is None:
|
||||
pending.append((target, int(previous) if previous is not None else default))
|
||||
for key, value in pending:
|
||||
params.put(key, value, block=True)
|
||||
params.put_bool('CancelButtonControlsMigrated', True, block=True)
|
||||
return True
|
||||
@@ -0,0 +1,79 @@
|
||||
"""Exact optional secondary Cancel profiles; no unsupported alternative bits."""
|
||||
import pytest
|
||||
from opendbc.car.gm.cancel_remap import supported, enabled, aol_enabled, valid_alternative_experience, prepare
|
||||
from opendbc.car.gm.tests.test_bolt_pedal import params
|
||||
from opendbc.car.gm.values import CAR
|
||||
|
||||
IDS = (CAR.CHEVROLET_BOLT_CC_2017, CAR.CHEVROLET_BOLT_CC_2018_2021,
|
||||
CAR.CHEVROLET_BOLT_CC_2022_2023, CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('candidate', IDS)
|
||||
@pytest.mark.parametrize('removed', [False, True])
|
||||
@pytest.mark.parametrize('aol', [False, True])
|
||||
def test_exact_configuration_keeps_optional_default_off(candidate, removed, aol):
|
||||
cp = params(candidate, pedal=True, camera=True, removed=removed)
|
||||
cp.alternativeExperience = 32 if aol else 0
|
||||
before = cp.as_reader().to_dict()
|
||||
assert supported(cp) and not enabled(cp) and valid_alternative_experience(cp)
|
||||
prepare(cp, True)
|
||||
assert cp.alternativeExperience == (96 if aol else 64)
|
||||
assert enabled(cp) and aol_enabled(cp) is aol and valid_alternative_experience(cp)
|
||||
after = cp.as_reader().to_dict()
|
||||
after['alternativeExperience'] = before['alternativeExperience']
|
||||
assert after == before
|
||||
prepare(cp, False)
|
||||
assert cp.alternativeExperience == (32 if aol else 0)
|
||||
assert not enabled(cp)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('mutation', ['stock', 'no_pedal', 'wrong_word', 'unknown_bits', 'passive'])
|
||||
def test_unqualified_configuration_cannot_remap(mutation):
|
||||
cp = params(CAR.CHEVROLET_BOLT_CC_2018_2021, pedal=mutation != 'no_pedal', camera=True)
|
||||
if mutation == 'stock':
|
||||
cp.openpilotLongitudinalControl = False
|
||||
elif mutation == 'wrong_word':
|
||||
cp.safetyConfigs[0].safetyParam = 0
|
||||
elif mutation == 'passive':
|
||||
cp.passive = True
|
||||
cp.alternativeExperience = 96 | (1 if mutation == 'unknown_bits' else 0)
|
||||
assert not valid_alternative_experience(cp)
|
||||
assert not enabled(cp)
|
||||
|
||||
|
||||
def test_assignment_migration_keeps_explicit_values_and_runs_once():
|
||||
from tempfile import TemporaryDirectory
|
||||
from openpilot.common.params import Params
|
||||
from opendbc.car.gm.cancel_remap import migrate_assignments
|
||||
cp = params(CAR.CHEVROLET_BOLT_CC_2018_2021, pedal=True, camera=True)
|
||||
prepare(cp, True)
|
||||
with TemporaryDirectory() as directory:
|
||||
saved = Params(directory)
|
||||
saved.put('CancelButtonControl', 8, block=True)
|
||||
saved.put('LongDistanceButtonControl', 3, block=True)
|
||||
assert migrate_assignments(saved, cp)
|
||||
assert saved.get('CancelButtonControl') == 8
|
||||
assert saved.get('LongCancelButtonControl') == 3
|
||||
assert saved.get('VeryLongCancelButtonControl') == 6
|
||||
saved.put('LongDistanceButtonControl', 4, block=True)
|
||||
assert not migrate_assignments(saved, cp)
|
||||
assert saved.get('LongCancelButtonControl') == 3
|
||||
|
||||
|
||||
@pytest.mark.parametrize('remap', [False, True])
|
||||
def test_cancel_experimental_assignment_does_not_invalidate_unrelated_map_for_exact_owner(remap):
|
||||
from tempfile import TemporaryDirectory
|
||||
from openpilot.common.params import Params
|
||||
from openpilot.starpilot.conditional_mode.manual import read_button_map
|
||||
from openpilot.starpilot.conditional_mode.button_actions import capture_sources, runtime_map_ready
|
||||
cp = params(CAR.CHEVROLET_BOLT_CC_2018_2021, pedal=True, camera=True)
|
||||
cp.alternativeExperience = 32
|
||||
prepare(cp, remap)
|
||||
with TemporaryDirectory() as directory:
|
||||
saved = Params(directory)
|
||||
saved.put('LongCancelButtonControl', 5, block=True)
|
||||
assert read_button_map(saved) is None
|
||||
assert (read_button_map(saved, cp=cp) is not None) is remap
|
||||
sources = capture_sources(saved, None, None)
|
||||
assert not runtime_map_ready(sources)
|
||||
assert runtime_map_ready(sources, cp=cp) is remap
|
||||
@@ -8,5 +8,6 @@ class ALTERNATIVE_EXPERIENCE:
|
||||
DISABLE_STOCK_AEB = 2
|
||||
RAISE_LONGITUDINAL_LIMITS_TO_ISO_MAX = 8
|
||||
ALLOW_AEB = 16
|
||||
GM_REMAP_CANCEL_TO_DISTANCE = 64
|
||||
TOYOTA_AUTO_HOLD = 128
|
||||
TOYOTA_AEB_HOLD = 256
|
||||
|
||||
@@ -204,8 +204,12 @@ static void gm_emit_paddle_after_stock(uint32_t now_us, uint32_t addr, uint8_t d
|
||||
if (gm_paddle_sched && feed->valid) {
|
||||
const bool expired = safety_get_ts_elapsed(now_us, feed->last_feed_us) > 100000U;
|
||||
if (expired) { feed->valid = false; } // Retain the command through its bounded host-feed lifetime.
|
||||
const bool applied = (addr == 0xBDU) ? (feed->bytes[0] == 0x20U) :
|
||||
((feed->bytes[3] & 0xFU) == (gm_bolt_gen2 ? 5U : 7U)) && (feed->bytes[5] == 2U);
|
||||
const bool main_ready = (!(gm_pedal_acc || gm_cancel_remap_enabled) || gm_pedal_main_ready());
|
||||
const bool inactive_release = gm_cancel_remap_enabled && !applied;
|
||||
if (!expired && gm_pedal_regen_gear_ready() && !regen_braking &&
|
||||
(!gm_pedal_acc || gm_pedal_main_ready())) {
|
||||
(main_ready || inactive_release)) {
|
||||
CANPacket_t packet = {0};
|
||||
packet.addr = addr;
|
||||
packet.bus = 0U;
|
||||
@@ -390,7 +394,7 @@ static void gm_rx_hook(const CANPacket_t *msg) {
|
||||
}
|
||||
|
||||
// exit controls on cancel press
|
||||
if (button == GM_BTN_CANCEL) {
|
||||
if ((button == GM_BTN_CANCEL) && !gm_cancel_remap_enabled) {
|
||||
controls_allowed = false;
|
||||
}
|
||||
|
||||
@@ -421,7 +425,7 @@ static void gm_rx_hook(const CANPacket_t *msg) {
|
||||
if ((msg_matches(msg, 0xC9U, 0U)) && (((gm_hw == GM_CAM) && !gm_camera_gateway) || (gm_camera_gateway && !gm_camera_pedal_f1)) && (!gm_ascm_intercept || gm_ascm_brake_c9) &&
|
||||
(!gm_sdgm || gm_sdgm_brake_c9)) {
|
||||
brake_pressed = GET_BIT(msg, 40U);
|
||||
if (gm_pedal_acc) {
|
||||
if ((gm_pedal_acc || (gm_cancel_remap_enabled && gm_pedal_long)) && (GET_LEN(msg) == 8U)) {
|
||||
acc_main_on = GET_BIT(msg, 29U);
|
||||
gm_pedal_main_seen = true;
|
||||
gm_pedal_main_last_us = microsecond_timer_get();
|
||||
@@ -611,6 +615,10 @@ static bool gm_tx_hook(const CANPacket_t *msg) {
|
||||
|
||||
bool steer_req = GET_BIT(msg, 3U);
|
||||
|
||||
if (gm_cancel_remap_enabled && ((desired_torque != 0) || steer_req) && !gm_pedal_main_ready()) {
|
||||
tx = false;
|
||||
}
|
||||
|
||||
if ((gm_volt_cc_long && ((desired_torque != 0) || steer_req)) || (gm_ordinary_cc_long && (desired_torque != 0))) { tx &= gm_volt_cc_sources_current() && gm_volt_cc_main && (gm_volt_cc_active || gm_aol_lateral_allowed()) && gm_volt_cc_forward_gear; }
|
||||
if (steer_torque_cmd_checks(desired_torque, steer_req, gm_bolt_2017 ? GM_BOLT_2017_STEERING_LIMITS : GM_STEERING_LIMITS)) {
|
||||
tx = false;
|
||||
@@ -664,7 +672,7 @@ static bool gm_tx_hook(const CANPacket_t *msg) {
|
||||
const int pair_delta = track1 - (2 * track2);
|
||||
const bool inactive = !enabled && (track1 == 0) && (track2 == 0);
|
||||
const bool active = enabled && gm_pedal_sensor_current() && gm_pedal_owns_longitudinal() && gm_pedal_drive_ready() &&
|
||||
(!gm_pedal_acc || gm_pedal_main_ready()) &&
|
||||
(!(gm_pedal_acc || gm_cancel_remap_enabled) || gm_pedal_main_ready()) &&
|
||||
get_longitudinal_allowed() && !brake_pressed_prev && !regen_braking &&
|
||||
(track1 >= 604) && (track1 <= 2633) && (track2 >= 304) && (track2 <= 1316) &&
|
||||
(pair_delta >= -16) && (pair_delta <= 16);
|
||||
@@ -682,7 +690,7 @@ static bool gm_tx_hook(const CANPacket_t *msg) {
|
||||
const bool applied = msg->data[0] == 0x20U;
|
||||
for (uint8_t i = 1U; i < 7U; i++) { if (msg->data[i] != 0U) { tx = false; } }
|
||||
if (!shape || (applied && !(gm_pedal_sensor_current() && gm_pedal_owns_longitudinal() &&
|
||||
(!gm_pedal_acc || gm_pedal_main_ready()) &&
|
||||
(!(gm_pedal_acc || gm_cancel_remap_enabled) || gm_pedal_main_ready()) &&
|
||||
gm_pedal_regen_gear_ready() && !regen_braking && get_longitudinal_allowed()))) {
|
||||
tx = false;
|
||||
}
|
||||
@@ -703,7 +711,7 @@ static bool gm_tx_hook(const CANPacket_t *msg) {
|
||||
(msg->data[3] == prndl) && (msg->data[4] == 0U) && (msg->data[6] == 1U) &&
|
||||
(msg->data[7] == 0U) && (applied || released);
|
||||
if (!shape || (applied && !(gm_pedal_sensor_current() && gm_pedal_owns_longitudinal() &&
|
||||
(!gm_pedal_acc || gm_pedal_main_ready()) &&
|
||||
(!(gm_pedal_acc || gm_cancel_remap_enabled) || gm_pedal_main_ready()) &&
|
||||
gm_pedal_regen_gear_ready() && !regen_braking && get_longitudinal_allowed()))) {
|
||||
tx = false;
|
||||
}
|
||||
|
||||
@@ -3,6 +3,14 @@
|
||||
static void gm_cruise_status_clear(void);
|
||||
|
||||
#define GM_ALT_EXP_ALWAYS_ON_LATERAL 32U
|
||||
#define GM_ALT_EXP_REMAP_CANCEL_TO_DISTANCE 64U
|
||||
|
||||
static bool gm_cancel_remap_enabled = false;
|
||||
|
||||
static bool gm_cancel_remap_word(uint16_t word) {
|
||||
return (word == 0xBDU) || (word == 0x9DU) || (word == 0x19DU) || (word == 0x1CDU) ||
|
||||
(word == 0xE700U) || (word == 0xE701U) || (word == 0xE702U) || (word == 0xE703U);
|
||||
}
|
||||
#define GM_AOL_MAIN_TIMEOUT_US 300000U
|
||||
|
||||
static bool gm_aol_enabled = false;
|
||||
@@ -44,7 +52,7 @@ static uint8_t gm_aol_permission_mask(void) {
|
||||
if (main_current && gear_current) {
|
||||
permission = request & 0x1U;
|
||||
}
|
||||
if (!gm_aol_stock_only && controls_allowed && ((request & 0x2U) != 0U)) {
|
||||
if (!gm_aol_stock_only && controls_allowed && (!gm_cancel_remap_enabled || main_current) && ((request & 0x2U) != 0U)) {
|
||||
permission |= 0x2U;
|
||||
}
|
||||
}
|
||||
@@ -79,7 +87,8 @@ static void gm_aol_observe(const CANPacket_t *msg) {
|
||||
// The caller supplies exact profile acceptance after its normal mode initialization.
|
||||
static void gm_aol_initialize(bool profile_supported) {
|
||||
gm_aol_reset();
|
||||
gm_aol_enabled = profile_supported && ((unsigned int)alternative_experience == GM_ALT_EXP_ALWAYS_ON_LATERAL);
|
||||
gm_aol_enabled = profile_supported && (((unsigned int)alternative_experience == GM_ALT_EXP_ALWAYS_ON_LATERAL) ||
|
||||
(gm_cancel_remap_enabled && ((unsigned int)alternative_experience == (GM_ALT_EXP_ALWAYS_ON_LATERAL | GM_ALT_EXP_REMAP_CANCEL_TO_DISTANCE))));
|
||||
if (gm_aol_enabled) {
|
||||
static const AolSafetyPolicy gm_aol_policy = {
|
||||
.reset = gm_aol_reset,
|
||||
|
||||
@@ -213,6 +213,14 @@ static unsigned int gm_bolt_cc_tag(uint16_t word) {
|
||||
static bool gm_bolt_cc_selected;
|
||||
|
||||
static safety_config gm_bolt_cc_init(uint16_t word) {
|
||||
const unsigned int experience = (unsigned int)alternative_experience;
|
||||
gm_cancel_remap_enabled = ((experience == GM_ALT_EXP_REMAP_CANCEL_TO_DISTANCE) ||
|
||||
(experience == (GM_ALT_EXP_ALWAYS_ON_LATERAL | GM_ALT_EXP_REMAP_CANCEL_TO_DISTANCE))) && gm_cancel_remap_word(word);
|
||||
if (((experience & GM_ALT_EXP_REMAP_CANCEL_TO_DISTANCE) != 0U) && !gm_cancel_remap_enabled) {
|
||||
gm_bolt_cc_selected = false;
|
||||
gm_hybrid_selected = false;
|
||||
return nooutput_init(0U);
|
||||
}
|
||||
gm_hybrid_rejected = ((word & 0xFF00U) == 0xE800U) && ((word > 0xE805U) ||
|
||||
((alternative_experience != 0) && (alternative_experience != 32)));
|
||||
gm_hybrid_selected = (word >= 0xE800U) && (word <= 0xE805U);
|
||||
|
||||
@@ -0,0 +1,165 @@
|
||||
"""Known Dom bit64 is admitted only for exact supported pedal-long Bolt words."""
|
||||
import pytest
|
||||
from opendbc.safety.tests.libsafety import libsafety_py
|
||||
from opendbc.car.gm.tests.test_bolt_cc import native
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def restore_native_configuration():
|
||||
safety = libsafety_py.libsafety
|
||||
safety.set_alternative_experience(0)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
safety.set_alternative_experience(0)
|
||||
assert safety.set_safety_hooks(19, 0) == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize('word', [0, 5, 16, 0xE710, 0xE711, 0xE712, 0xE713, 0xC140, 0xE800])
|
||||
@pytest.mark.parametrize('experience', [64, 96])
|
||||
def test_remap_bit_on_unqualified_profile_has_no_native_output(word, experience):
|
||||
safety = libsafety_py.libsafety
|
||||
safety.init_tests()
|
||||
safety.set_alternative_experience(experience)
|
||||
assert safety.set_safety_hooks(4, word) == 0
|
||||
safety.set_aol_test_heartbeat(True)
|
||||
safety.aol_set_host_request(3)
|
||||
safety.set_controls_allowed(True)
|
||||
assert safety.aol_get_permission_mask() == 0
|
||||
assert not native('tx', (0x180, bytes(4), 0), 1_000_000)
|
||||
assert not native('tx', (0x200, bytes(6), 0), 1_000_000)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('experience', [65, 66, 97, 98, 127, 255])
|
||||
@pytest.mark.parametrize('word', [0xBD, 0x9D, 0x19D, 0x1CD, 0xE700, 0xE701, 0xE702, 0xE703])
|
||||
def test_known_profile_does_not_allow_unknown_bits_with_remap(word, experience):
|
||||
safety = libsafety_py.libsafety
|
||||
safety.init_tests()
|
||||
safety.set_alternative_experience(experience)
|
||||
assert safety.set_safety_hooks(4, word) == 0
|
||||
safety.set_aol_test_heartbeat(True)
|
||||
safety.aol_set_host_request(3)
|
||||
safety.set_controls_allowed(True)
|
||||
assert safety.aol_get_permission_mask() == 0
|
||||
assert not native('tx', (0x180, bytes(4), 0), 1_000_000)
|
||||
assert not native('tx', (0x200, bytes(6), 0), 1_000_000)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('word', [0x9D, 0x1CD, 0xE701, 0xE703])
|
||||
@pytest.mark.parametrize('experience', [64, 96])
|
||||
@pytest.mark.parametrize('condition', ['on', 'absent', 'off', 'stale'])
|
||||
def test_remap_requires_fresh_physical_main_for_all_active_outputs(word, experience, condition):
|
||||
from opendbc.can import CANPacker
|
||||
from opendbc.car import Bus
|
||||
from opendbc.car.gm.gmcan import create_pedal_command, create_steering_control, create_bolt_regen_paddle, create_bolt_regen_gear
|
||||
from opendbc.car.gm.values import CAR, DBC
|
||||
from opendbc.car.gm.tests.test_cc_gateway_stock import pt_frames
|
||||
from opendbc.car.gm.tests.test_bolt_pedal import TestBoltPedalMessages
|
||||
safety = libsafety_py.libsafety
|
||||
gen2 = word in (0x1CD, 0xE703)
|
||||
packer = CANPacker(DBC[CAR.CHEVROLET_BOLT_CC_2018_2021][Bus.pt])
|
||||
safety.init_tests()
|
||||
safety.set_alternative_experience(experience)
|
||||
assert safety.set_safety_hooks(4, word) == 0
|
||||
safety.reset_recorded_can()
|
||||
now = 1_000_000
|
||||
|
||||
def feed(stamp, main, counter):
|
||||
safety.set_timer(stamp)
|
||||
packets = pt_frames(packer, main=main is not False, cruise=False, counter=counter % 4, acc_cruise=0)
|
||||
packets = [packet for packet in packets if packet[0] not in (0x1F5, 0xC9)]
|
||||
if main is not None:
|
||||
packets.append(packer.make_can_msg('ECMEngineStatus', 0, {'CruiseMainOn': int(main)}))
|
||||
packets.extend([packer.make_can_msg('ECMPRDNL2', 0, {'PRNDL2': 6}),
|
||||
packer.make_can_msg('EBCMRegenPaddle', 0, {'RegenPaddle': 0}),
|
||||
TestBoltPedalMessages.sensor(packer, 0., counter % 16),
|
||||
(0x3D1, bytes(8), 0)])
|
||||
for packet in packets:
|
||||
assert native('rx', packet, stamp)
|
||||
safety.set_aol_test_heartbeat(True)
|
||||
if experience == 96:
|
||||
safety.aol_set_host_request(3)
|
||||
feed(now, None if condition == 'absent' else True, 0)
|
||||
feed(now + 10_000, None if condition == 'absent' else True, 1)
|
||||
safety.set_controls_allowed(True) # Established full engagement before the Main transition.
|
||||
if condition == 'off':
|
||||
feed(now + 20_000, False, 2)
|
||||
assert not safety.get_controls_allowed()
|
||||
elif condition == 'stale':
|
||||
# Refresh every other physical source so only the original Main receipt expires.
|
||||
feed(now + 310_001, None, 2)
|
||||
now += 310_001 if condition == 'stale' else 20_000
|
||||
safety.set_timer(now)
|
||||
safety.set_torque_driver(0, 0)
|
||||
safety.set_desired_torque_last(1)
|
||||
safety.set_rt_torque_last(1)
|
||||
expected = condition == 'on'
|
||||
assert bool(native('tx', create_steering_control(packer, 0, 1, 0, True), now)) is expected
|
||||
assert bool(native('tx', create_pedal_command(packer, .2, 1), now)) is expected
|
||||
assert native('tx', create_steering_control(packer, 0, 0, 1, False), now)
|
||||
assert native('tx', create_pedal_command(packer, 0., 2), now)
|
||||
# Host paddle feeds are never sent directly. Stock RX invokes the checked scheduler.
|
||||
safety.reset_recorded_can()
|
||||
assert not native('tx', create_bolt_regen_paddle(packer, True), now)
|
||||
assert not native('tx', create_bolt_regen_gear(packer, True, gen2), now)
|
||||
native('rx', packer.make_can_msg('EBCMRegenPaddle', 0, {'RegenPaddle': 0}), now)
|
||||
native('rx', packer.make_can_msg('ECMPRDNL2', 0, {'PRNDL2': 6}), now)
|
||||
assert bool(safety.get_recorded_can_count()) is expected
|
||||
if experience == 96 and not expected:
|
||||
assert safety.aol_get_permission_mask() == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize('word', [0x9D, 0x1CD, 0xE701, 0xE703])
|
||||
@pytest.mark.parametrize('experience', [64, 96])
|
||||
@pytest.mark.parametrize('condition', ['off', 'stale'])
|
||||
def test_previously_accepted_applied_paddle_feed_cannot_outlive_main(word, experience, condition):
|
||||
from opendbc.can import CANPacker
|
||||
from opendbc.car import Bus
|
||||
from opendbc.car.gm.gmcan import create_bolt_regen_paddle, create_bolt_regen_gear
|
||||
from opendbc.car.gm.values import CAR, DBC
|
||||
from opendbc.car.gm.tests.test_cc_gateway_stock import pt_frames
|
||||
from opendbc.car.gm.tests.test_bolt_pedal import TestBoltPedalMessages
|
||||
safety = libsafety_py.libsafety
|
||||
gen2 = word in (0x1CD, 0xE703)
|
||||
packer = CANPacker(DBC[CAR.CHEVROLET_BOLT_CC_2018_2021][Bus.pt])
|
||||
safety.init_tests()
|
||||
safety.set_alternative_experience(experience)
|
||||
assert safety.set_safety_hooks(4, word) == 0
|
||||
|
||||
def feed(stamp, main, counter):
|
||||
packets = pt_frames(packer, cruise=False, counter=counter % 4, acc_cruise=0)
|
||||
packets = [packet for packet in packets if packet[0] not in (0xC9, 0x1F5)]
|
||||
packets.extend([packer.make_can_msg('ECMPRDNL2', 0, {'PRNDL2': 6}),
|
||||
packer.make_can_msg('EBCMRegenPaddle', 0, {'RegenPaddle': 0}),
|
||||
TestBoltPedalMessages.sensor(packer, 0., counter % 16), (0x3D1, bytes(8), 0)])
|
||||
if main is not None:
|
||||
packets.append(packer.make_can_msg('ECMEngineStatus', 0, {'CruiseMainOn': int(main)}))
|
||||
for packet in packets:
|
||||
assert native('rx', packet, stamp)
|
||||
safety.set_aol_test_heartbeat(True)
|
||||
if experience == 96:
|
||||
safety.aol_set_host_request(3)
|
||||
feed(1_000_000, True, 0)
|
||||
# Main remains fresh at 280ms; the host feed lease remains fresh when Main later expires.
|
||||
feed(1_280_000, None, 1)
|
||||
safety.set_controls_allowed(True)
|
||||
assert safety.aol_get_permission_mask() == (3 if experience == 96 else 0)
|
||||
assert not native('tx', create_bolt_regen_paddle(packer, True), 1_280_000)
|
||||
assert not native('tx', create_bolt_regen_gear(packer, True, gen2), 1_280_000)
|
||||
# Prove the same feed actually emits while Main is fresh before testing withdrawal.
|
||||
safety.reset_recorded_can()
|
||||
native('rx', packer.make_can_msg('EBCMRegenPaddle', 0, {'RegenPaddle': 0}), 1_280_001)
|
||||
native('rx', packer.make_can_msg('ECMPRDNL2', 0, {'PRNDL2': 6}), 1_280_001)
|
||||
assert safety.get_recorded_can_count() == 2
|
||||
withdrawn = 1_290_000 if condition == 'off' else 1_310_001
|
||||
feed(withdrawn, False if condition == 'off' else None, 2)
|
||||
safety.reset_recorded_can()
|
||||
native('rx', packer.make_can_msg('EBCMRegenPaddle', 0, {'RegenPaddle': 0}), withdrawn + 1)
|
||||
native('rx', packer.make_can_msg('ECMPRDNL2', 0, {'PRNDL2': 6}), withdrawn + 1)
|
||||
assert safety.get_recorded_can_count() == 0
|
||||
# A new inactive feed can still release the synthetic paddle while Main is absent.
|
||||
assert not native('tx', create_bolt_regen_paddle(packer, False), withdrawn + 2)
|
||||
assert not native('tx', create_bolt_regen_gear(packer, False, gen2), withdrawn + 2)
|
||||
native('rx', packer.make_can_msg('EBCMRegenPaddle', 0, {'RegenPaddle': 0}), withdrawn + 3)
|
||||
native('rx', packer.make_can_msg('ECMPRDNL2', 0, {'PRNDL2': 6}), withdrawn + 3)
|
||||
assert safety.get_recorded_can_count() == 2
|
||||
@@ -150,6 +150,8 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
|
||||
{"StarButtonControl", {PERSISTENT, INT, "0"}},
|
||||
{"LongStarButtonControl", {PERSISTENT, INT, "0"}},
|
||||
{"VeryLongStarButtonControl", {PERSISTENT, INT, "0"}},
|
||||
{"CancelButtonControlsMigrated", {PERSISTENT, BOOL, "0"}},
|
||||
{"RemapCancelToDistance", {PERSISTENT, BOOL, "0"}},
|
||||
{"CancelButtonControl", {PERSISTENT, INT, "0"}},
|
||||
{"LongCancelButtonControl", {PERSISTENT, INT, "0"}},
|
||||
{"VeryLongCancelButtonControl", {PERSISTENT, INT, "0"}},
|
||||
|
||||
@@ -159,7 +159,8 @@ class CarEvents:
|
||||
for b in CS.buttonEvents:
|
||||
# Disable on rising and falling edge of cancel for both stock and OP long
|
||||
# TODO: only check the cancel button with openpilot longitudinal on all brands to match panda safety
|
||||
if b.type == ButtonType.cancel and (allow_button_cancel or not self.CP.pcmCruise):
|
||||
from opendbc.car.gm.cancel_remap import enabled as gm_cancel_remap_enabled
|
||||
if b.type == ButtonType.cancel and not gm_cancel_remap_enabled(self.CP) and (allow_button_cancel or not self.CP.pcmCruise):
|
||||
events.add(EventName.buttonCancel)
|
||||
|
||||
# Handle permanent and temporary steering faults
|
||||
|
||||
@@ -308,7 +308,7 @@ class Car:
|
||||
self.switchback_button_tracker = ButtonTracker() if self.switchback_capable else None
|
||||
self.switchback_settings_owner = ConditionalSettingsOwner(self.params) if self.switchback_capable else None
|
||||
self.gm_distance = GMDistanceButtons() if self.conditional_replay and gm_profiles_supported(self.CP) else None
|
||||
self.gm_distance_map = WheelMapCache(include_ioniq_media=False) if self.gm_distance is not None else None
|
||||
self.gm_distance_map = WheelMapCache(include_ioniq_media=False, cp=self.CP) if self.gm_distance is not None else None
|
||||
self.gm_distance_claim_tracker = ButtonTracker() if self.gm_distance is not None else None
|
||||
controller_available = self.CI.CC is not None and openpilot_enabled_toggle and not self.CP.dashcamOnly
|
||||
self.CP.passive = not controller_available or self.CP.dashcamOnly
|
||||
@@ -318,6 +318,8 @@ class Car:
|
||||
self.CP.safetyConfigs = [safety_config]
|
||||
|
||||
startup_preferences.finalize(self.CP)
|
||||
from opendbc.car.gm.cancel_remap import migrate_assignments as migrate_gm_cancel_assignments
|
||||
migrate_gm_cancel_assignments(self.params, self.CP)
|
||||
startup_preferences.configure_controller(self.CI)
|
||||
configure_controller(self.CI, self.params)
|
||||
self.vehicle_startup.configure(self.CI)
|
||||
@@ -481,7 +483,7 @@ class Car:
|
||||
if (getattr(self.CI.CS, 'ev6_aol_source_healthy', False) or
|
||||
qualified_honda_stock_aol(self.CP, marked_only=True) or
|
||||
qualified_gm(self.CP) and is_bolt_present_no_acc_pedal_profile(self.CP) and
|
||||
int(self.CP.alternativeExperience) == 32) else 0)
|
||||
int(self.CP.alternativeExperience) in (32, 96)) else 0)
|
||||
self.observe_ioniq6_long_authority(can_list, CS)
|
||||
media_owner = getattr(self, 'ioniq6_media', None)
|
||||
media_observation = media_owner.update(can_list) if media_owner is not None else None
|
||||
|
||||
@@ -303,7 +303,7 @@ class SelfdriveD:
|
||||
self.events.add(EventName.belowSteerSpeed)
|
||||
paddle_pressed = paddle_cancel(self.CP, CS, enabled=self.enabled, saved=self.nostalgia_enabled)
|
||||
self.nostalgia_paddle_cancel |= bool(paddle_pressed and EventName.buttonCancel not in self.events.names and
|
||||
not physical_cancel(CS))
|
||||
not physical_cancel(CS, self.CP))
|
||||
if paddle_pressed:
|
||||
self.events.add(EventName.buttonCancel)
|
||||
|
||||
@@ -549,7 +549,7 @@ class SelfdriveD:
|
||||
self.event_car_state = CS
|
||||
if reconcile_input:
|
||||
self._update_steering_authority(CS)
|
||||
if any(physical_cancel(sample) for sample, _ in samples):
|
||||
if any(physical_cancel(sample, self.CP) for sample, _ in samples):
|
||||
self.nostalgia_paddle_cancel = False
|
||||
if CS.canValid:
|
||||
if self.tesla_stock_consumer.binding is not None:
|
||||
@@ -884,7 +884,7 @@ class SelfdriveD:
|
||||
if any(not valid or not sample.canValid or sample.canTimeout or sample.steerFaultTemporary or
|
||||
sample.steerFaultPermanent or (sample.brakePressed or sample.regenBraking) and not held_brake or
|
||||
active_longitudinal and sample.gasPressed and not held_gas or
|
||||
sample.seatbeltUnlatched or sample.accFaulted or sample.doorOpen or physical_cancel(sample) or
|
||||
sample.seatbeltUnlatched or sample.accFaulted or sample.doorOpen or physical_cancel(sample, self.CP) or
|
||||
sample.gearShifter not in (car.CarState.GearShifter.drive, car.CarState.GearShifter.sport,
|
||||
car.CarState.GearShifter.low, car.CarState.GearShifter.brake)
|
||||
for sample, valid in samples):
|
||||
|
||||
@@ -4,13 +4,14 @@ Register divergence remains a protected-register diagnostic, not a benign-fault
|
||||
claim. Raw fault reports are retained. Every unlisted configuration/bit blocks.
|
||||
"""
|
||||
from openpilot.cereal import log
|
||||
from opendbc.car.gm.cancel_remap import aol_enabled
|
||||
|
||||
|
||||
def _diagnostic_mask(cp):
|
||||
try:
|
||||
from opendbc.car.gm.aol import qualified_gm
|
||||
from opendbc.car.gm.values import CAR, is_volt_sdgm_profile, is_bolt_present_no_acc_pedal_profile
|
||||
if cp.brand != 'gm' or not qualified_gm(cp) or int(cp.alternativeExperience) != 32 or len(cp.safetyConfigs) != 1:
|
||||
if cp.brand != 'gm' or not qualified_gm(cp) or not aol_enabled(cp) or len(cp.safetyConfigs) != 1:
|
||||
return 0
|
||||
word = int(cp.safetyConfigs[0].safetyParam)
|
||||
if (cp.carFingerprint == CAR.CHEVROLET_VOLT_2019 and word == 0x1005 and
|
||||
|
||||
@@ -71,7 +71,7 @@ def stock_lateral_fault_proof(self, CS, message, now_ns):
|
||||
samples = getattr(self, 'car_state_samples', ()) or ((CS, True),)
|
||||
if any(not valid or not sample.canValid or sample.canTimeout or sample.steerFaultPermanent or
|
||||
sample.steerFaultTemporary or sample.accFaulted or sample.doorOpen or sample.seatbeltUnlatched or
|
||||
sample.brakePressed or sample.regenBraking or sample.gasPressed or physical_cancel(sample) or
|
||||
sample.brakePressed or sample.regenBraking or sample.gasPressed or physical_cancel(sample, self.CP) or
|
||||
sample.gearShifter not in (car.CarState.GearShifter.drive, car.CarState.GearShifter.sport,
|
||||
car.CarState.GearShifter.low, car.CarState.GearShifter.brake) for sample, valid in samples):
|
||||
return None
|
||||
@@ -162,7 +162,7 @@ def ordinary_source_availability(self, CS, now_ns, native=None, *, replay: bool,
|
||||
bool(getattr(self, 'car_state_retained_valid', False) or held_metadata)),)
|
||||
if any(not valid or not sample.canValid or sample.canTimeout or sample.steerFaultPermanent or
|
||||
sample.steerFaultTemporary or sample.accFaulted or sample.doorOpen or sample.seatbeltUnlatched or
|
||||
sample.brakePressed or sample.regenBraking or sample.gasPressed or physical_cancel(sample) or
|
||||
sample.brakePressed or sample.regenBraking or sample.gasPressed or physical_cancel(sample, self.CP) or
|
||||
sample.gearShifter not in (car.CarState.GearShifter.drive, car.CarState.GearShifter.sport,
|
||||
car.CarState.GearShifter.low, car.CarState.GearShifter.brake)
|
||||
for sample, valid in samples):
|
||||
@@ -269,7 +269,7 @@ def fresh_ordinary_lag_continuation(self, CS, intent, native, now_ns):
|
||||
samples = getattr(self, 'car_state_samples', ()) or ((CS, True),)
|
||||
return not any(not valid or not sample.canValid or sample.canTimeout or sample.steerFaultPermanent or
|
||||
sample.steerFaultTemporary or sample.accFaulted or sample.doorOpen or sample.seatbeltUnlatched or
|
||||
sample.brakePressed or sample.regenBraking or sample.gasPressed or physical_cancel(sample) or
|
||||
sample.brakePressed or sample.regenBraking or sample.gasPressed or physical_cancel(sample, self.CP) or
|
||||
sample.gearShifter not in (car.CarState.GearShifter.drive, car.CarState.GearShifter.sport,
|
||||
car.CarState.GearShifter.low, car.CarState.GearShifter.brake)
|
||||
for sample, valid in samples)
|
||||
|
||||
@@ -0,0 +1,233 @@
|
||||
"""Actual parser/native/ordinary event boundaries for optional GM Cancel mapping."""
|
||||
from contextlib import ExitStack
|
||||
import pytest
|
||||
from opendbc.car.gm.tests.test_bolt_cc import native
|
||||
from opendbc.car.gm.tests.test_bolt_pedal import params
|
||||
from opendbc.car.gm.values import CAR
|
||||
from opendbc.safety.tests.libsafety import libsafety_py
|
||||
from openpilot.cereal import log
|
||||
from openpilot.common.params import Params
|
||||
from openpilot.common.prefix import OpenpilotPrefix
|
||||
from openpilot.selfdrive.car.car_events import CarEvents
|
||||
from openpilot.starpilot.aol.tests.test_gm import TestGmAol as GmCardFixture
|
||||
from openpilot.starpilot.aol.tests.test_gm_held_native import NOW, parse as source_fixture
|
||||
from openpilot.starpilot.nostalgia import physical_cancel
|
||||
|
||||
IDS = (CAR.CHEVROLET_BOLT_CC_2017, CAR.CHEVROLET_BOLT_CC_2018_2021,
|
||||
CAR.CHEVROLET_BOLT_CC_2022_2023, CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def restore_native_configuration():
|
||||
safety = libsafety_py.libsafety
|
||||
safety.set_alternative_experience(0)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
safety.set_alternative_experience(0)
|
||||
assert safety.set_safety_hooks(19, 0) == 0
|
||||
|
||||
|
||||
def parse(card, now, gesture=None, counter=0, *, belt_unlatched=False, packets_only=False):
|
||||
from opendbc.can import CANPacker
|
||||
from opendbc.car import Bus
|
||||
from opendbc.car.gm.values import DBC
|
||||
packets = source_fixture(card, now, gesture, counter, belt_unlatched=belt_unlatched, packets_only=True)
|
||||
if card.CP.carFingerprint == CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL:
|
||||
packer = CANPacker(DBC[card.CP.carFingerprint][Bus.pt])
|
||||
packets.append(packer.make_can_msg('ASCMActiveCruiseControlStatus', 2, {'ACCCmdActive': 0}))
|
||||
if packets_only:
|
||||
return packets
|
||||
if now == NOW:
|
||||
card.CI.update([(now - 1_000_000, packets)])
|
||||
observed = card.CI.update([(now, packets)])
|
||||
assert observed.canValid and not observed.canTimeout
|
||||
assert observed.brakePressed == (gesture == 'brake')
|
||||
assert observed.regenBraking == (gesture == 'paddle')
|
||||
assert observed.seatbeltUnlatched == belt_unlatched
|
||||
card._aol_can_observation = (observed, now, now, now)
|
||||
return observed
|
||||
|
||||
|
||||
@pytest.mark.parametrize('candidate', IDS)
|
||||
@pytest.mark.parametrize('remap', [False, True])
|
||||
def test_actual_card_parser_native_and_event_cancel_contract(candidate, remap):
|
||||
with ExitStack() as cleanup:
|
||||
cleanup.enter_context(OpenpilotPrefix())
|
||||
saved = Params()
|
||||
saved.put_bool('AlwaysOnLateral', True, block=True)
|
||||
saved.put_bool('RemapCancelToDistance', remap, block=True)
|
||||
cp = params(candidate, pedal=True, camera=True)
|
||||
card = GmCardFixture.card(cp, saved)
|
||||
cleanup.callback(card.vehicle_startup.close)
|
||||
assert card.CP.alternativeExperience == (96 if remap else 32)
|
||||
safety = libsafety_py.libsafety
|
||||
safety.init_tests()
|
||||
safety.set_alternative_experience(card.CP.alternativeExperience)
|
||||
assert safety.set_safety_hooks(4, card.CP.safetyConfigs[0].safetyParam) == 0
|
||||
for tick in range(10):
|
||||
now = NOW + tick * 10_000_000
|
||||
for packet in parse(card, now, counter=tick % 4, packets_only=True):
|
||||
assert native('rx', packet, now // 1000)
|
||||
parse(card, now, counter=tick % 4)
|
||||
safety.set_aol_test_heartbeat(True)
|
||||
safety.aol_set_host_request(3)
|
||||
safety.set_controls_allowed(True)
|
||||
assert safety.aol_get_permission_mask() == 3
|
||||
cancel_packet = (0x1E1, bytes.fromhex('00000000016490'), 0)
|
||||
assert native('rx', cancel_packet, (NOW + 95_000_000) // 1000)
|
||||
assert bool(safety.get_controls_allowed()) is remap
|
||||
assert safety.aol_get_permission_mask() == (3 if remap else 1)
|
||||
previous = parse(card, NOW + 100_000_000, counter=2)
|
||||
observed = parse(card, NOW + 110_000_000, gesture='cancel', counter=3)
|
||||
# Raw parser input stays visible for the existing secondary-action tracker.
|
||||
assert any(str(event.type) == 'cancel' for event in observed.buttonEvents)
|
||||
assert physical_cancel(observed, card.CP) is (not remap)
|
||||
events = CarEvents(card.CP).update(observed, previous, card.CC_prev)
|
||||
assert (log.OnroadEvent.EventName.buttonCancel in events.names) is (not remap)
|
||||
card.aol_card_intent.update(previous, fault_active=False, now_ns=NOW + 100_000_000)
|
||||
assert card.aol_card_intent.allowed_latch
|
||||
card.aol_card_intent.update(observed, fault_active=False, now_ns=NOW + 110_000_000)
|
||||
assert card.aol_card_intent.allowed_latch
|
||||
# The remap never exempts a real regen withdrawal.
|
||||
assert native('rx', (0xBD, bytes.fromhex('20000000000000'), 0), (NOW + 115_000_000) // 1000)
|
||||
assert not safety.get_controls_allowed() and safety.aol_get_permission_mask() == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize('candidate', IDS)
|
||||
@pytest.mark.parametrize('remap', [False, True])
|
||||
def test_actual_selfdrived_controls_preserve_full_op_only_for_configured_remap(candidate, remap):
|
||||
from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
|
||||
from opendbc.car.gm.cancel_remap import prepare
|
||||
case = fixture.IoniqAvailabilityTests(methodName='runTest')
|
||||
with ExitStack() as cleanup:
|
||||
cleanup.callback(case.doCleanups)
|
||||
cleanup.enter_context(OpenpilotPrefix())
|
||||
saved = Params()
|
||||
saved.put_bool('RemapCancelToDistance', remap, block=True)
|
||||
cp = params(candidate, pedal=True, camera=True)
|
||||
prepare(cp, remap)
|
||||
sd, _, step = case._authority_callers(True, cp=cp, genuine_events=True)
|
||||
card = GmCardFixture.card(cp, saved)
|
||||
cleanup.callback(card.vehicle_startup.close)
|
||||
for tick in range(10):
|
||||
parse(card, NOW + tick * 10_000_000, counter=tick % 4)
|
||||
axis, state, command = step(tick)
|
||||
assert state.enabled and axis.lateralActive and axis.longitudinalActive
|
||||
assert command.latActive and command.longActive
|
||||
observed = parse(card, NOW + 100_000_000, gesture='cancel', counter=2)
|
||||
axis, state, command = step(10, observed_state=observed, allowed=True,
|
||||
native_request_lat=True, native_request_long=True, long_allowed=remap,
|
||||
panda_controls_allowed=remap)
|
||||
assert bool(state.enabled) is remap
|
||||
assert bool(axis.longitudinalActive) is remap and bool(command.longActive) is remap
|
||||
assert log.OnroadEvent.EventName.controlsMismatch not in sd.events.names
|
||||
if remap:
|
||||
assert axis.lateralActive and command.latActive
|
||||
|
||||
|
||||
@pytest.mark.parametrize('candidate', [IDS[1], IDS[3]])
|
||||
@pytest.mark.parametrize('remap', [False, True])
|
||||
@pytest.mark.parametrize('slot,frames', [(0, 1), (1, 50), (2, 250)])
|
||||
def test_actual_card_parser_dispatches_auxiliary_action_only_when_remapped(candidate, remap, slot, frames):
|
||||
from openpilot.starpilot.aol.intent import read_settings
|
||||
with ExitStack() as cleanup:
|
||||
cleanup.enter_context(OpenpilotPrefix())
|
||||
saved = Params()
|
||||
saved.put_bool('AlwaysOnLateral', True, block=True)
|
||||
saved.put_bool('RemapCancelToDistance', remap, block=True)
|
||||
keys = ('CancelButtonControl', 'LongCancelButtonControl', 'VeryLongCancelButtonControl')
|
||||
for index, key in enumerate(keys):
|
||||
saved.put(key, 3 if index == slot else 0, block=True)
|
||||
card = GmCardFixture.card(params(candidate, pedal=True, camera=True), saved)
|
||||
cleanup.callback(card.vehicle_startup.close)
|
||||
owner = card.aol_card_intent
|
||||
assert owner.settings == read_settings(saved)
|
||||
neutral = parse(card, NOW, counter=0)
|
||||
owner.update_auxiliary(neutral)
|
||||
for tick in range(frames):
|
||||
observed = parse(card, NOW + (tick + 1) * 10_000_000, gesture='cancel', counter=(tick + 1) % 4)
|
||||
before = observed.to_dict()
|
||||
owner.update_auxiliary(observed)
|
||||
assert observed.to_dict() == before
|
||||
release = parse(card, NOW + (frames + 1) * 10_000_000, counter=(frames + 1) % 4)
|
||||
owner.update_auxiliary(release)
|
||||
assert owner.pause_lateral is remap
|
||||
|
||||
|
||||
@pytest.mark.parametrize('candidate', [IDS[1], IDS[3]])
|
||||
@pytest.mark.parametrize('remap', [False, True])
|
||||
@pytest.mark.parametrize('slot,frames', [(0, 1), (1, 50), (2, 250)])
|
||||
def test_actual_card_parser_and_wheel_publisher_dispatch_once(candidate, remap, slot, frames):
|
||||
from openpilot.starpilot.controllers.wheel_actions import WheelPublisher
|
||||
with ExitStack() as cleanup:
|
||||
cleanup.enter_context(OpenpilotPrefix())
|
||||
saved = Params()
|
||||
saved.put_bool('RemapCancelToDistance', remap, block=True)
|
||||
keys = ('CancelButtonControl', 'LongCancelButtonControl', 'VeryLongCancelButtonControl')
|
||||
for index, key in enumerate(keys):
|
||||
saved.put(key, 8 if index == slot else 0, block=True)
|
||||
card = GmCardFixture.card(params(candidate, pedal=True, camera=True), saved)
|
||||
cleanup.callback(card.vehicle_startup.close)
|
||||
publisher = WheelPublisher(saved)
|
||||
commands = []
|
||||
drive_id = NOW - 1_000_000
|
||||
neutral = parse(card, NOW, counter=0)
|
||||
publisher.observe(saved, card.CP, neutral, now_ns=NOW, drive_id=drive_id)
|
||||
for tick in range(frames):
|
||||
now = NOW + (tick + 1) * 10_000_000
|
||||
observed = parse(card, now, gesture='cancel', counter=(tick + 1) % 4)
|
||||
commands.extend(publisher.observe(saved, card.CP, observed, now_ns=now, drive_id=drive_id))
|
||||
now = NOW + (frames + 1) * 10_000_000
|
||||
released = parse(card, now, counter=(frames + 1) % 4)
|
||||
commands.extend(publisher.observe(saved, card.CP, released, now_ns=now, drive_id=drive_id))
|
||||
actions = [(key, action) for key, action in commands if key]
|
||||
assert actions == ([(keys[slot], 8)] if remap else [])
|
||||
assert all((key, action) == ('', 0) for key, action in commands if not key)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('candidate', IDS)
|
||||
@pytest.mark.parametrize('remap', [False, True])
|
||||
def test_actual_card_parser_native_and_event_cancel_contract_aol_off(candidate, remap):
|
||||
"""Ordinary Card/native operation has no AOL owner or request at experience0/64."""
|
||||
with ExitStack() as cleanup:
|
||||
cleanup.enter_context(OpenpilotPrefix())
|
||||
saved = Params()
|
||||
saved.put_bool('AlwaysOnLateral', False, block=True)
|
||||
saved.put_bool('RemapCancelToDistance', remap, block=True)
|
||||
cp = params(candidate, pedal=True, camera=True)
|
||||
card = GmCardFixture.card(cp, saved)
|
||||
cleanup.callback(card.vehicle_startup.close)
|
||||
assert not saved.get_bool('AlwaysOnLateral')
|
||||
assert card.CP.alternativeExperience == (64 if remap else 0)
|
||||
assert not card.aol_qualified and card.aol_card_intent is None
|
||||
safety = libsafety_py.libsafety
|
||||
safety.init_tests()
|
||||
safety.set_alternative_experience(card.CP.alternativeExperience)
|
||||
assert safety.set_safety_hooks(4, card.CP.safetyConfigs[0].safetyParam) == 0
|
||||
for tick in range(10):
|
||||
now = NOW + tick * 10_000_000
|
||||
for packet in parse(card, now, counter=tick % 4, packets_only=True):
|
||||
assert native('rx', packet, now // 1000)
|
||||
parse(card, now, counter=tick % 4)
|
||||
# The established ordinary full-engagement starting condition is explicit;
|
||||
# unlike the AOL cases, this path never issues an AOL host request.
|
||||
safety.set_controls_allowed(True)
|
||||
assert safety.aol_get_request_mask() == 0
|
||||
assert safety.aol_get_permission_mask() == 0
|
||||
cancel_packet = (0x1E1, bytes.fromhex('00000000016490'), 0)
|
||||
assert native('rx', cancel_packet, (NOW + 95_000_000) // 1000)
|
||||
assert bool(safety.get_controls_allowed()) is remap
|
||||
assert safety.aol_get_request_mask() == 0
|
||||
assert safety.aol_get_permission_mask() == 0
|
||||
previous = parse(card, NOW + 100_000_000, counter=2)
|
||||
observed = parse(card, NOW + 110_000_000, gesture='cancel', counter=3)
|
||||
assert any(str(event.type) == 'cancel' for event in observed.buttonEvents)
|
||||
assert physical_cancel(observed, card.CP) is (not remap)
|
||||
events = CarEvents(card.CP).update(observed, previous, card.CC_prev)
|
||||
assert (log.OnroadEvent.EventName.buttonCancel in events.names) is (not remap)
|
||||
assert card.aol_card_intent is None
|
||||
assert native('rx', (0xBD, bytes.fromhex('20000000000000'), 0), (NOW + 115_000_000) // 1000)
|
||||
assert not safety.get_controls_allowed()
|
||||
assert safety.aol_get_request_mask() == 0
|
||||
assert safety.aol_get_permission_mask() == 0
|
||||
@@ -111,7 +111,7 @@ def stock_lateral_off(cp, cs, native):
|
||||
return bool(cp.pcmCruise and cs.cruiseState.available and not cs.cruiseState.enabled and
|
||||
cs.canValid and not cs.canTimeout and not cs.steerFaultPermanent and not cs.steerFaultTemporary and
|
||||
not cs.accFaulted and not cs.doorOpen and not cs.seatbeltUnlatched and
|
||||
not cs.brakePressed and not cs.regenBraking and not cs.gasPressed and not physical_cancel(cs) and
|
||||
not cs.brakePressed and not cs.regenBraking and not cs.gasPressed and not physical_cancel(cs, cp) and
|
||||
native is not None and native.requestedLateral and native.lateralAllowed and not native.requestedLongitudinal)
|
||||
|
||||
|
||||
@@ -286,7 +286,7 @@ class TransportPauseFeedback:
|
||||
not 0 < self.lateral_source_ns <= axis.sourceCarStateMonoTime <= axis.observedMonoTime <= now_ns or
|
||||
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
|
||||
not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or cs.accFaulted or
|
||||
cs.steeringDisengage or physical_cancel(cs) or
|
||||
cs.steeringDisengage or physical_cancel(cs, cp) or
|
||||
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
|
||||
not 0 < int(sm.logMonoTime['onroadEvents']) <= now_ns or
|
||||
now_ns - int(sm.logMonoTime['onroadEvents']) > 1_500_000_000 or
|
||||
@@ -305,7 +305,7 @@ class TransportPauseFeedback:
|
||||
def _temporary_host_intent_current(self, sm, cp, cs, axis, stamp, now_ns, *, latched, producer):
|
||||
"""A fresh inactive proof may retain intent, never native/output authority."""
|
||||
from openpilot.starpilot.nostalgia import physical_cancel
|
||||
if physical_cancel(cs) or cs.steeringDisengage:
|
||||
if physical_cancel(cs, cp) or cs.steeringDisengage:
|
||||
return False
|
||||
events = sm['onroadEvents']
|
||||
faults = [event for event in events if event.immediateDisable or event.softDisable]
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
"""GM-owned AOL capabilities; exact CP admission lives with the vehicle."""
|
||||
|
||||
import math
|
||||
from opendbc.car.gm.cancel_remap import aol_enabled, enabled as cancel_remap_enabled, supported as cancel_remap_supported
|
||||
|
||||
from opendbc.car.structs import CarParams
|
||||
from opendbc.car.gm.values import CAR, control_flags, camera_acc_pedal_profile, is_volt_ascm_longitudinal
|
||||
@@ -14,7 +15,7 @@ def policy_for(cp) -> AolVehiclePolicy:
|
||||
return AolVehiclePolicy(intent_supported=qualified, settings_supported=qualified,
|
||||
runtime_supported=qualified, normal_runtime_supported=qualified,
|
||||
lkas_button_supported=False,
|
||||
full_axis_runtime_required=qualified and int(cp.alternativeExperience) == GM_AOL_ALTERNATIVE_EXPERIENCE,
|
||||
full_axis_runtime_required=qualified and aol_enabled(cp),
|
||||
alternative_experience_addition=GM_AOL_ALTERNATIVE_EXPERIENCE if qualified else 0)
|
||||
|
||||
|
||||
@@ -23,7 +24,7 @@ def native_profile_supported(model: int, param: int) -> bool:
|
||||
|
||||
|
||||
def native_accepts_cp(cp, model: int, param: int) -> bool:
|
||||
return (qualified_gm(cp) and int(cp.alternativeExperience) == GM_AOL_ALTERNATIVE_EXPERIENCE and
|
||||
return (qualified_gm(cp) and aol_enabled(cp) and
|
||||
native_profile_supported(model, param))
|
||||
|
||||
|
||||
@@ -40,6 +41,14 @@ class GmAolCardIntent(AolCardIntent):
|
||||
self.startup_calibration_pending = True
|
||||
self._startup_incomplete = False
|
||||
|
||||
def update_auxiliary(self, CS, *, media=None, media_eligible=False, fault_active=None):
|
||||
semantic = CS
|
||||
if (cancel_remap_supported(self.cp) and not cancel_remap_enabled(self.cp) and
|
||||
any(str(event.type) == 'cancel' for event in CS.buttonEvents)):
|
||||
semantic = CS.as_builder() if hasattr(CS, 'as_builder') else CS.as_reader().as_builder()
|
||||
semantic.buttonEvents = [event for event in CS.buttonEvents if str(event.type) != 'cancel']
|
||||
super().update_auxiliary(semantic, media=media, media_eligible=media_eligible, fault_active=fault_active)
|
||||
|
||||
@property
|
||||
def main_action(self):
|
||||
# GM MAIN reports a persistent physical level, not an independent AOL toggle.
|
||||
@@ -209,7 +218,7 @@ def physical_longitudinal_withdrawal_source(cp, cs, parser, observation, now_ns)
|
||||
from opendbc.car.structs import car
|
||||
try:
|
||||
if (not qualified_gm(cp) or not is_bolt_present_no_acc_pedal_profile(cp) or
|
||||
int(cp.alternativeExperience) != GM_AOL_ALTERNATIVE_EXPERIENCE or
|
||||
not aol_enabled(cp) or
|
||||
observation is None or observation[0] is not cs or not cs.canValid or cs.canTimeout):
|
||||
return 0
|
||||
fields = []
|
||||
@@ -243,7 +252,7 @@ def retained_physical_longitudinal_intent(feedback, sm, cp, cs, axis, now_ns, *,
|
||||
from openpilot.starpilot.aol.transport_pause import independent_lateral_health
|
||||
occurrence = feedback.gm_withdrawal_ns or source_ns
|
||||
if (not qualified_gm(cp) or not is_bolt_present_no_acc_pedal_profile(cp) or
|
||||
int(cp.alternativeExperience) != GM_AOL_ALTERNATIVE_EXPERIENCE or
|
||||
not aol_enabled(cp) or
|
||||
not latched or feedback.failed or feedback.paused or feedback.pending_event or
|
||||
feedback.retention_started_ns or not feedback.session or str(axis.sessionId) != feedback.session or
|
||||
not feedback.healthy_feedback_ns or not 0 <= now_ns - feedback.healthy_feedback_ns <= AXIS_MAX_AGE_NS or
|
||||
@@ -296,7 +305,7 @@ def physical_cancel_lateral_allowed(cp, cs, intent, native, *, source_ns, now_ns
|
||||
from openpilot.starpilot.aol.runtime import AXIS_MAX_AGE_NS, SAFETY_MAX_AGE_NS, native_inventory_matches_cp
|
||||
try:
|
||||
return bool(qualified_gm(cp) and is_bolt_present_no_acc_pedal_profile(cp) and
|
||||
int(cp.alternativeExperience) == GM_AOL_ALTERNATIVE_EXPERIENCE and
|
||||
aol_enabled(cp) and
|
||||
intent is not None and intent.settingsQualified and intent.allowedLatch and intent.lateralArmed and
|
||||
intent.producerSessionId and intent.carStateLogMonoTime == source_ns and
|
||||
0 < source_ns <= intent.observedMonoTime <= now_ns <= intent.validUntilMonoTime and
|
||||
@@ -322,7 +331,7 @@ def seatbelt_lateral_soft_event_allowed(cp, cs, intent, native, event, *, source
|
||||
from openpilot.starpilot.aol.runtime import AXIS_MAX_AGE_NS, SAFETY_MAX_AGE_NS, native_inventory_matches_cp
|
||||
try:
|
||||
return bool(is_bolt_present_no_acc_pedal_profile(cp) and
|
||||
int(cp.alternativeExperience) == GM_AOL_ALTERNATIVE_EXPERIENCE and
|
||||
aol_enabled(cp) and
|
||||
event.name == log.OnroadEvent.EventName.seatbeltNotLatched and event.softDisable and
|
||||
event.noEntry and not event.immediateDisable and not event.userDisable and
|
||||
cs.seatbeltUnlatched and cs.canValid and not cs.canTimeout and not cs.accFaulted and not cs.doorOpen and
|
||||
|
||||
@@ -81,12 +81,13 @@ def display_action(raw: bytes | None) -> tuple[str, tuple[str, ...], str]:
|
||||
return "Unsupported saved action", (), "Off"
|
||||
|
||||
|
||||
def runtime_map_ready(sources: ButtonSources) -> bool:
|
||||
def runtime_map_ready(sources: ButtonSources, *, cp=None) -> bool:
|
||||
"""Mirror the runtime map's scalar admission using the already captured bytes."""
|
||||
if not sources.readable or any(raw is not None and raw not in _CANONICAL_ACTIONS
|
||||
for _, raw in sources.dependencies[2:]):
|
||||
return False
|
||||
return all(sources.raw(key) != b"5" for key in
|
||||
from opendbc.car.gm.cancel_remap import enabled as gm_cancel_remap_enabled
|
||||
return gm_cancel_remap_enabled(cp) or all(sources.raw(key) != b"5" for key in
|
||||
("CancelButtonControl", "LongCancelButtonControl", "VeryLongCancelButtonControl"))
|
||||
|
||||
|
||||
|
||||
@@ -60,7 +60,7 @@ def conditional_manual_candidate(CS: car.CarState, tracker: ButtonTracker, param
|
||||
if not media.valid:
|
||||
tracker.observe_media(media)
|
||||
elif media.samples:
|
||||
buttons = media_buttons if media_map_captured else read_button_map(params, include_ioniq_media=True)
|
||||
buttons = media_buttons if media_map_captured else read_button_map(params, include_ioniq_media=True, cp=CP)
|
||||
if buttons is None:
|
||||
tracker.invalidate_media()
|
||||
return None
|
||||
@@ -73,9 +73,9 @@ def conditional_manual_candidate(CS: car.CarState, tracker: ButtonTracker, param
|
||||
return None
|
||||
if gesture.button in (Button.LKAS, Button.DISTANCE):
|
||||
# Read the saved mapping on each physical gesture so live remaps apply without 0x448.
|
||||
buttons = read_button_map(params, include_ioniq_media=ioniq_media)
|
||||
buttons = read_button_map(params, include_ioniq_media=ioniq_media, cp=CP)
|
||||
elif buttons is None:
|
||||
buttons = media_buttons if media_map_captured and ioniq_media else read_button_map(params, include_ioniq_media=ioniq_media)
|
||||
buttons = media_buttons if media_map_captured and ioniq_media else read_button_map(params, include_ioniq_media=ioniq_media, cp=CP)
|
||||
if buttons is None or not buttons.assigned(gesture):
|
||||
return None
|
||||
return gesture, drive_id, fingerprint, choice, now_ns
|
||||
@@ -130,7 +130,7 @@ def conditional_traffic_candidate(CS: car.CarState, tracker: ButtonTracker, para
|
||||
if tracker.media_context is not None and tracker.media_context[:2] != (drive_id, fingerprint):
|
||||
tracker.invalidate_media()
|
||||
return None
|
||||
buttons = media_buttons if media_map_captured else read_button_map(params, include_ioniq_media=not gm_distance)
|
||||
buttons = media_buttons if media_map_captured else read_button_map(params, include_ioniq_media=not gm_distance, cp=CP)
|
||||
if buttons is None:
|
||||
tracker.invalidate_media()
|
||||
return None
|
||||
|
||||
@@ -102,8 +102,8 @@ def _action(params, key: str) -> int | None:
|
||||
return int(raw)
|
||||
|
||||
|
||||
def read_button_map(params, *, include_ioniq_media: bool = False) -> ButtonMap | None:
|
||||
"""Read registered slots; frozen GM-only cancel remap stays inert here."""
|
||||
def read_button_map(params, *, include_ioniq_media: bool = False, cp=None) -> ButtonMap | None:
|
||||
"""Read non-Cancel slots; the exact GM owner dispatches remapped Cancel separately."""
|
||||
keys = ('LKASButtonControl', 'DistanceButtonControl', 'LongDistanceButtonControl',
|
||||
'VeryLongDistanceButtonControl', 'CancelButtonControl', 'LongCancelButtonControl',
|
||||
'VeryLongCancelButtonControl')
|
||||
@@ -115,7 +115,8 @@ def read_button_map(params, *, include_ioniq_media: bool = False) -> ButtonMap |
|
||||
return None
|
||||
assert all(value is not None for value in values)
|
||||
# Cancel-to-distance is GM Pedal-only; this gesture source must not reinterpret cancel.
|
||||
if EXPERIMENTAL_MODE_ACTION in values[4:7]:
|
||||
from opendbc.car.gm.cancel_remap import enabled as gm_cancel_remap_enabled
|
||||
if EXPERIMENTAL_MODE_ACTION in values[4:7] and not gm_cancel_remap_enabled(cp):
|
||||
return None
|
||||
return ButtonMap(*(cast(int, value) for value in (*values[:4], *values[7:])))
|
||||
|
||||
@@ -128,8 +129,9 @@ class WheelMapCache:
|
||||
consumer can interpret that source packet.
|
||||
"""
|
||||
|
||||
def __init__(self, *, include_ioniq_media=True):
|
||||
def __init__(self, *, include_ioniq_media=True, cp=None):
|
||||
self.include_ioniq_media = include_ioniq_media
|
||||
self.cp = cp
|
||||
self.buttons: ButtonMap | None = None
|
||||
self.checked_ns = 0
|
||||
self.held: tuple[bool, ...] | None = None
|
||||
@@ -169,7 +171,7 @@ class WheelMapCache:
|
||||
self.last_sample_ns = sample.source_boot_ns
|
||||
if (changed_epoch or self.checked_ns == 0 or edge or boundary or
|
||||
now_ns < self.checked_ns or now_ns - self.checked_ns >= 1_000_000_000):
|
||||
self.buttons = read_button_map(params, include_ioniq_media=self.include_ioniq_media)
|
||||
self.buttons = read_button_map(params, include_ioniq_media=self.include_ioniq_media, cp=self.cp)
|
||||
self.checked_ns = now_ns
|
||||
return self.buttons
|
||||
|
||||
|
||||
@@ -30,9 +30,10 @@ class ManualSettingsTests(unittest.TestCase):
|
||||
original = read_button_map
|
||||
calls = []
|
||||
|
||||
def read_once(source, *, include_ioniq_media=False):
|
||||
def read_once(source, *, include_ioniq_media=False, cp=None):
|
||||
self.assertIsNone(cp)
|
||||
calls.append(include_ioniq_media)
|
||||
return original(source, include_ioniq_media=include_ioniq_media)
|
||||
return original(source, include_ioniq_media=include_ioniq_media, cp=cp)
|
||||
|
||||
def packet(mode, custom, stamp, now):
|
||||
media = MediaObservation((MediaSample(mode, custom, stamp),), stamp, 0, True)
|
||||
|
||||
@@ -138,7 +138,7 @@ class TrafficOwner:
|
||||
if (self.cached_buttons is None or incoming_toggle or
|
||||
(incoming_map and incoming_map != self.map_fingerprint) or
|
||||
now_mono_ns - self.map_checked_ns >= 1_000_000_000):
|
||||
buttons = read_button_map(params, include_ioniq_media=not gm_distance)
|
||||
buttons = read_button_map(params, include_ioniq_media=not gm_distance, cp=cp)
|
||||
self.map_checked_ns = now_mono_ns
|
||||
if buttons is None:
|
||||
prior = self.requested
|
||||
|
||||
@@ -7,8 +7,9 @@ from openpilot.starpilot.conditional_mode.traffic_status import TrafficDisplayPr
|
||||
|
||||
|
||||
class TrafficLaunchState:
|
||||
def __init__(self, params):
|
||||
def __init__(self, params, *, cp=None):
|
||||
self.params = params
|
||||
self.cp = cp
|
||||
self.settings = ConditionalSettingsOwner(params)
|
||||
self.projector = TrafficDisplayProjector()
|
||||
self.buttons = None
|
||||
@@ -16,7 +17,7 @@ class TrafficLaunchState:
|
||||
|
||||
def sample(self, sm, *, now_ns, boot_ns, drive_id):
|
||||
if now_ns < self.checked_ns or now_ns - self.checked_ns >= 1_000_000_000:
|
||||
self.buttons = read_button_map(self.params)
|
||||
self.buttons = read_button_map(self.params, cp=self.cp)
|
||||
self.checked_ns = now_ns
|
||||
if self.buttons is None:
|
||||
return None
|
||||
|
||||
@@ -221,7 +221,7 @@ def apply_switchback_gesture(owner, event, *, params, settings, sm, cp, now_ns:
|
||||
snapshot = settings.refresh(now_ns)
|
||||
verdict = settings.verdict(snapshot, now_mono_ns=now_ns, drive_id=drive)
|
||||
fingerprint = settings_fingerprint(snapshot)
|
||||
buttons = read_button_map(params, include_ioniq_media=True)
|
||||
buttons = read_button_map(params, include_ioniq_media=True, cp=cp)
|
||||
if verdict.status != 'ready' or verdict.safe_mode is not False or fingerprint is None or buttons is None:
|
||||
return False
|
||||
session, sequence = str(wire.sessionId), int(wire.sequence)
|
||||
|
||||
@@ -138,6 +138,11 @@ class WheelPublisher:
|
||||
state, media if ioniq6_media_eligible(cp) else None)]
|
||||
for tracker, source, keys in ((self.cancel_tracker, structs.CarState.ButtonEvent.Type.cancel, KEYS[5:8]),
|
||||
(self.main_tracker, structs.CarState.ButtonEvent.Type.mainCruise, KEYS[1:2])):
|
||||
if source == structs.CarState.ButtonEvent.Type.cancel and cp.brand == 'gm':
|
||||
from opendbc.car.gm.cancel_remap import enabled as gm_cancel_remap_enabled, supported as gm_cancel_remap_supported
|
||||
if gm_cancel_remap_supported(cp) and not gm_cancel_remap_enabled(cp):
|
||||
self.cancel_tracker = ButtonTracker()
|
||||
continue
|
||||
if not any(mapping[key] in DELIVERED_ACTIONS for key in keys):
|
||||
continue
|
||||
copy = structs.CarState(canValid=state.canValid, canTimeout=state.canTimeout)
|
||||
|
||||
@@ -115,7 +115,7 @@ class LongitudinalInputs:
|
||||
if self.gm_start_enabled:
|
||||
self.gm_boot_offset_ns = None
|
||||
self.gm_source_floor_ns = 0
|
||||
self.gm_traffic_state = TrafficLaunchState(self.params)
|
||||
self.gm_traffic_state = TrafficLaunchState(self.params, cp=self.CP)
|
||||
self.gm_profile_host = ProfileHost(self.params) if feature_enabled(self.params, self.CP, 'profile', os.environ) else None
|
||||
if self.toyota_sienna_replay or self.toyota_sienna_production or self.toyota_corolla_stop:
|
||||
self.toyota_boot_offset_ns: int | None = None
|
||||
|
||||
@@ -23,7 +23,10 @@ def paddle_cancel(cp, cs, *, enabled: bool, saved: bool) -> bool:
|
||||
any(event.type == ButtonType.altButton2 and event.pressed for event in cs.buttonEvents))
|
||||
|
||||
|
||||
def physical_cancel(cs) -> bool:
|
||||
def physical_cancel(cs, cp=None) -> bool:
|
||||
from opendbc.car.gm.cancel_remap import enabled as gm_cancel_remap_enabled
|
||||
if cp is not None and gm_cancel_remap_enabled(cp):
|
||||
return False
|
||||
return any(event.type == ButtonType.cancel for event in cs.buttonEvents)
|
||||
|
||||
|
||||
|
||||
@@ -101,7 +101,7 @@ class ConditionalFeature:
|
||||
sources = capture_sources(self.owner.params, document, safe)
|
||||
enabled = (self.owner.authority("switchback_wheel" if switchback_only else "conditional_wheel") and config_valid and
|
||||
safe_readable and safe in (None, b"0", b"1") and sources.readable)
|
||||
ready = runtime_map_ready(sources) and all(display_action(sources.raw(key))[0] != "Unsupported saved action"
|
||||
ready = runtime_map_ready(sources, cp=cp) and all(display_action(sources.raw(key))[0] != "Unsupported saved action"
|
||||
for key in keys)
|
||||
rows = [] if ready else [FeatureRow("", "Button assignments",
|
||||
"Review saved actions",
|
||||
|
||||
@@ -75,7 +75,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,
|
||||
"LongPitch": True, "DisableOpenpilotLongitudinal": False,
|
||||
"RemapCancelToDistance": False, "LongPitch": True, "DisableOpenpilotLongitudinal": False,
|
||||
"SpeedLimitController": False, "ShowSpeedLimits": True,
|
||||
"SLCConfirmation": False,
|
||||
"PauseLateralOnSignal": False,
|
||||
@@ -151,6 +151,8 @@ class FeatureSettingsOwner:
|
||||
configuration_longitudinal=self.configuration_longitudinal)
|
||||
self.long_profiles = LongProfileFeature(self)
|
||||
self.tesla_screen = TeslaScreenFeature(self)
|
||||
from openpilot.starpilot.ui.gm_cancel_remap_feature import GmCancelRemapFeature
|
||||
self.gm_cancel_remap = GmCancelRemapFeature(self)
|
||||
self.gm_tune = GmTuneFeature(self)
|
||||
self.gm_truck = GmTruckFeature(self)
|
||||
self.gm_ev_preset = GmEvPresetFeature(self)
|
||||
@@ -737,6 +739,7 @@ class FeatureSettingsOwner:
|
||||
"native safety and ECU configuration are unchanged."),
|
||||
capability=radar_capability, default_value="Off"))
|
||||
rows.extend(self.tesla_screen.rows(parked))
|
||||
rows.extend(self.gm_cancel_remap.rows(parked))
|
||||
rows.extend(self.gm_tune.rows(parked))
|
||||
rows.extend(self.gm_truck.rows(parked))
|
||||
rows.extend(self.gm_ev_preset.rows(parked))
|
||||
@@ -1169,6 +1172,8 @@ class FeatureSettingsOwner:
|
||||
return self.controller.apply_setup(request)
|
||||
if key == 'ForceAutoTuneOff':
|
||||
return self.controller.apply_learning(request)
|
||||
if key == "RemapCancelToDistance":
|
||||
return self.gm_cancel_remap.apply(request)
|
||||
if key == "GMPedalLongitudinal":
|
||||
return self._apply_pedal_setup(request)
|
||||
if key in TESLA_SCREEN_KEYS:
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Saved next-drive secondary Cancel input on exact supported pedal-long Bolts."""
|
||||
import hashlib
|
||||
from dataclasses import replace
|
||||
from opendbc.car.gm.cancel_remap import KEY, supported
|
||||
from openpilot.starpilot.saved_document import commit_exact
|
||||
|
||||
|
||||
class GmCancelRemapFeature:
|
||||
def __init__(self, owner):
|
||||
self.owner = owner
|
||||
|
||||
def capability(self):
|
||||
cp = self.owner.vehicle_params()
|
||||
if not supported(cp):
|
||||
return None
|
||||
reader = cp.as_reader() if hasattr(cp, "as_reader") else cp
|
||||
return (str(cp.carFingerprint), hashlib.sha256(reader.as_builder().to_bytes()).hexdigest())
|
||||
|
||||
def rows(self, parked):
|
||||
capability = self.capability()
|
||||
if capability is None:
|
||||
return []
|
||||
row = self.owner._bool_row(KEY, "Use Cancel as Secondary Button", parked and self.owner.authority("parked_preferences"),
|
||||
"Use the Cancel button assignments in Wheel Controls instead of cancelling cruise. Applies next drive.")
|
||||
return [replace(row, capability=capability, available=row.available and self.owner._readable(KEY))]
|
||||
|
||||
def apply(self, request):
|
||||
capability = self.capability()
|
||||
if capability is None or request.capability != capability or request.value not in ("Off", "On"):
|
||||
return False
|
||||
def authorized():
|
||||
return self.owner.authority("parked_preferences") and self.capability() == capability
|
||||
result = commit_exact(self.owner.params, key=KEY, max_bytes=8, expected=request.expected,
|
||||
raw=b"1" if request.value == "On" else b"0", authorized=authorized, temp_prefix=".gm-cancel-")
|
||||
return result.committed and result.verified
|
||||
@@ -42,6 +42,10 @@ class WheelFeature:
|
||||
return ()
|
||||
if key in KEYS[:2] and policy.fixed_cruise_buttons:
|
||||
return ()
|
||||
if key in KEYS[5:8] and cp.brand == 'gm':
|
||||
from opendbc.car.gm.cancel_remap import supported
|
||||
if supported(cp) and read_saved(self.owner.params, 'RemapCancelToDistance', 8) != (b'1', True):
|
||||
return ()
|
||||
if key in KEYS[8:] and not media:
|
||||
return ()
|
||||
if key in KEYS[:5] and not policy.settings_supported:
|
||||
|
||||
@@ -24,6 +24,7 @@ class VehicleStartupPreferences:
|
||||
tesla_preap_stock: bool = True
|
||||
volt_sng: bool = False
|
||||
gm_auto_hold: bool = False
|
||||
gm_cancel_remap: bool = False
|
||||
volt_one_pedal: bool = False
|
||||
tesla_screen: bool = False
|
||||
tesla_screen_brake: bool = False
|
||||
@@ -87,6 +88,8 @@ class VehicleStartupPreferences:
|
||||
read_saved(params, "LockDoors", 8) == (b"1", True)),
|
||||
toyota_unlock_doors=bool(enabled and readable and safe in (None, b"0") and
|
||||
read_saved(params, "UnlockDoors", 8) == (b"1", True)),
|
||||
gm_cancel_remap=bool(enabled and not disable_bolt and readable and safe in (None, b"0") and
|
||||
read_saved(params, "RemapCancelToDistance", 8) == (b"1", True)),
|
||||
volt_sng=volt_sng, gm_auto_hold=gm_auto_hold, volt_one_pedal=volt_one_pedal,
|
||||
gm_camera_pedal=bool(enabled and not disable_bolt and readable and safe in (None, b"0")),
|
||||
toyota_filter=bool(enabled and not disable_bolt and readable and safe in (None, b"0")),
|
||||
@@ -154,6 +157,8 @@ class VehicleStartupPreferences:
|
||||
self._prepare_camera_pedal(cp)
|
||||
apply_gm_auto_hold(cp, self.gm_auto_hold)
|
||||
apply_volt_one_pedal(cp, self.volt_one_pedal and not self.disable_bolt_long, self.gm_auto_hold)
|
||||
from opendbc.car.gm.cancel_remap import prepare as prepare_cancel_remap
|
||||
prepare_cancel_remap(cp, self.gm_cancel_remap)
|
||||
if cp.brand == "toyota":
|
||||
apply_toyota_auto_hold(cp, self.toyota_auto_hold)
|
||||
from opendbc.car.toyota.ars510_preferences import apply_radar_mode
|
||||
@@ -179,6 +184,8 @@ class VehicleStartupPreferences:
|
||||
self._prepare_camera_pedal(cp)
|
||||
apply_gm_auto_hold(cp, self.gm_auto_hold and admitted_hold)
|
||||
apply_volt_one_pedal(cp, self.volt_one_pedal and admitted_one_pedal and not self.disable_bolt_long, self.gm_auto_hold and admitted_hold)
|
||||
from opendbc.car.gm.cancel_remap import prepare as prepare_cancel_remap
|
||||
prepare_cancel_remap(cp, self.gm_cancel_remap)
|
||||
if cp.brand == "toyota":
|
||||
admitted = bool(cp.flags & ToyotaFlags.AUTO_BRAKE_HOLD)
|
||||
apply_toyota_auto_hold(cp, self.toyota_auto_hold and admitted)
|
||||
|
||||
Reference in New Issue
Block a user