IQ.Lvbs/IQ.Pilot
8
7
Fork 3
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

IQ.Pilot Release Commit @ ab07000

Browse Source
This commit is contained in:
IQ.Lvbs CI [bot]
2026-08-08 11:48:21 -05:00
parent 31c07687c4
commit 9e52535231
42 changed files with 368 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
iqdbc_repo/iqdbc/car/volkswagen/carcontroller.py
View File

@@ -245,6 +245,9 @@ class CarController(CarControllerBase):
self.leadDistanceBars = 0
self.lead_distance_bars_last = None
self.distance_bar_frame = 0
self.mlb_hud_text = 0
self.mlb_hud_text_frame = 0
self.mlb_set_speed_last = 0
self.speed_limit_last = 0
self.speed_limit_changed_timer = 0
self.blinkerActive = None
@@ -280,6 +283,19 @@ class CarController(CarControllerBase):
return float(np.interp(v_ego, [0.4, 3.5, 4.0], [0.8, 0.95, 1.0]))
return 1.0
def _mlb_acc_hud_text(self, hud_control, set_speed: float) -> int:
# ACC_02 primary display text, briefly surfaced on a follow distance or set speed change
if hud_control.leadDistanceBars != self.lead_distance_bars_last:
self.mlb_hud_text_frame = self.frame
self.mlb_hud_text = self.CCP.ACC_HUD_TEXT_DISTANCE.get(hud_control.leadDistanceBars, self.CCP.ACC_HUD_TEXTS["none"])
elif set_speed != self.mlb_set_speed_last and hud_control.speedVisible:
self.mlb_hud_text_frame = self.frame
self.mlb_hud_text = self.CCP.ACC_HUD_TEXTS["setSpeed"]
elif self.frame - self.mlb_hud_text_frame >= self.CCP.ACC_HUD_TEXT_STEP:
self.mlb_hud_text = self.CCP.ACC_HUD_TEXTS["none"]
self.mlb_set_speed_last = set_speed
return self.mlb_hud_text
def _should_spam_mqb_a0_resume(self, CS, enabled: bool) -> bool:
return bool(
enabled and
@@ -429,8 +445,9 @@ class CarController(CarControllerBase):
can_sends.append(self.CCS.create_blinker_control(self.packer_pt, self.CAN.pt, CS.ea_hud_stock_values, CS.ea_control_stock_values,
left_blinker, right_blinker, self.hide_ea_error))
if self.CP.openpilotLongitudinalControl and self.CCS == mqbcan and not self.acc_counter_seeded and CS.acc_stock_counters:
for name in ("ACC_02", "ACC_06", "ACC_07", "ACC_10"):
if self.CP.openpilotLongitudinalControl and self.CCS in (mqbcan, mlbcan) and not self.acc_counter_seeded and CS.acc_stock_counters:
seed_msgs = ("ACC_01", "ACC_02") if self.CCS is mlbcan else ("ACC_02", "ACC_06", "ACC_07", "ACC_10")
for name in seed_msgs:
addr = self.packer_pt.dbc.name_to_msg[name].address
self.packer_pt.counters[addr] = (CS.acc_stock_counters[name] + 1) % 16
self.acc_counter_seeded = True
@@ -499,6 +516,8 @@ class CarController(CarControllerBase):
self.long_deviation, self.long_jerklimit, eBrakeActive,
esp_starting_override=esp_starting_override, esp_stopping_override=esp_stopping_override,
))
elif self.CCS == mlbcan:
can_sends.extend(self.CCS.create_acc_accel_control(self.packer_pt, self.CAN.pt, accel, acc_control, stopping))
else:
accel = apply_pq_stopping_accel(self.CP.carFingerprint, accel, stopping)
@@ -577,7 +596,10 @@ class CarController(CarControllerBase):
acc_hud_status = self.CCS.acc_hud_status_value(CS.out.cruiseState.available, CS.out.accFaulted, CC.longActive, CC.cruiseControl.override)
set_speed = hud_control.setSpeed * CV.MS_TO_KPH
decel = dVisual(self.CCS, CS)
can_sends.append(self.CCS.create_acc_hud_control(self.packer_pt, self.CAN.pt, acc_hud_status, set_speed, leadDistance, self.leadDistanceBars, fcw_alert, hud_control.leadVisible, self.unavailable, decel, d_unresponsive))
hud_kwargs = {"hud_text": self._mlb_acc_hud_text(hud_control, set_speed)} if self.CCS is mlbcan else {}
can_sends.append(self.CCS.create_acc_hud_control(self.packer_pt, self.CAN.pt, acc_hud_status, set_speed, leadDistance,
self.leadDistanceBars, fcw_alert, hud_control.leadVisible, self.unavailable,
decel, d_unresponsive, **hud_kwargs))
if self.CP.flags & VolkswagenFlags.PQ:
iq_lvbs_commander.update_turn_signals(self, CC, CS, can_sends)

22
iqdbc_repo/iqdbc/car/volkswagen/carstate.py
View File

@@ -675,9 +675,12 @@ class CarState(CarStateBase):
else:
ret.gearShifter = GearShifter.drive
# ACC okay but disabled (1), ACC ready (2), a radar visibility or other fault/disruption (6 or 7)
# currently regulating speed (3), driver accel override (4), brake only (5)
if self.CP.carFingerprint == CAR.PORSCHE_MACAN_MK1:
cruise_main_switch = bool(pt_cp.vl["LS_01"]["LS_Hauptschalter"])
if not self.CP.pcmCruise:
ret.cruiseState.available = cruise_main_switch
ret.cruiseState.enabled = False
ret.accFaulted = False
elif self.CP.carFingerprint == CAR.PORSCHE_MACAN_MK1:
ret.cruiseState.available = ext_cp.vl["ACC_05"]["ACC_Status_ACC"] in (2, 3, 4, 5)
ret.cruiseState.enabled = ext_cp.vl["ACC_05"]["ACC_Status_ACC"] in (3, 4, 5)
ret.accFaulted = ext_cp.vl["ACC_05"]["ACC_Status_ACC"] in (6, 7)
@@ -687,6 +690,12 @@ class CarState(CarStateBase):
ret.cruiseState.speed = ext_cp.vl["ACC_02"]["ACC_Wunschgeschw_02"] * CV.KPH_TO_MS
ret.accFaulted = pt_cp.vl["TSK_02"]["TSK_Status"] in (3,)
ret.cruiseState.nonAdaptive = bool(pt_cp.vl["LS_01"]["LS_Limiter"])
if not self.CP.pcmCruise:
self.acc_stock_counters["ACC_01"] = int(ext_cp.vl["ACC_01"]["COUNTER"])
self.acc_stock_counters["ACC_02"] = int(ext_cp.vl["ACC_02"]["COUNTER"])
self.esp_hold_confirmation = bool(pt_cp.vl["ESP_02"]["ESP_Stillstandsflag"])
self.parse_mlb_mqb_steering_state(ret, pt_cp)
self._update_mlb_iq_alc_state(pt_cp)
@@ -728,9 +737,10 @@ class CarState(CarStateBase):
ret.cruiseState.standstill = self.CP.pcmCruise and self.esp_hold_confirmation
ret.standstill = ret.vEgoRaw == 0
ret.cruiseFaultLateralMode = False
ret.lateralAvailable = ret.cruiseState.available
ret.blockPcmEnable = False
allow_lat_only = self._params.get_bool("AllowLateralWhenLongUnavailable") and self._params.get_bool("AolEnabled")
ret.cruiseFaultLateralMode = allow_lat_only and ret.accFaulted and cruise_main_switch
ret.lateralAvailable = ret.cruiseState.available or ret.cruiseFaultLateralMode
ret.blockPcmEnable = ret.cruiseFaultLateralMode
self.cruise_faulted = ret.accFaulted
self.frame += 1

1
iqdbc_repo/iqdbc/car/volkswagen/interface.py
View File

@@ -91,6 +91,7 @@ class CarInterface(CarInterfaceBase):
safety_configs = [get_safety_config(structs.CarParams.SafetyModel.volkswagenMlb)]
ret.enableBsm = 0x30F in fingerprint[0] # SWA_01
ret.networkLocation = NetworkLocation.gateway
ret.transmissionType = TransmissionType.automatic
ret.dashcamOnly = False
elif ret.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO):

65
iqdbc_repo/iqdbc/car/volkswagen/mlbcan.py
View File

@@ -1,4 +1,11 @@
from iqdbc.car.volkswagen.mqbcan import (volkswagen_mqb_meb_checksum, xor_checksum, create_lka_hud_control as mqb_create_lka_hud_control)
from iqdbc.car.volkswagen.mqbcan import (volkswagen_mqb_meb_checksum, xor_checksum,
acc_control_value as mqb_acc_control_value,
acc_hud_status_value as mqb_acc_hud_status_value,
create_lka_hud_control as mqb_create_lka_hud_control)
# ACC_01.ACC_Sollbeschleunigung one increment above range max, "no acceleration request"
ACC_INACTIVE_ACCEL = 3.01
def create_hca_steering_control(packer, bus, apply_steer, HCA_Status):
values = {
@@ -11,8 +18,10 @@ def create_hca_steering_control(packer, bus, apply_steer, HCA_Status):
return packer.make_can_msg("HCA_01", bus, values)
def create_lka_hud_control(packer, bus, ldw_stock_values, enabled, steering_pressed, hud_alert, hud_control, entering=False, special_mode=False, special_active=False):
return mqb_create_lka_hud_control(packer, bus, ldw_stock_values, enabled, steering_pressed, hud_alert, hud_control, entering, special_mode, special_active)
def create_lka_hud_control(packer, bus, ldw_stock_values, enabled, steering_pressed, hud_alert, hud_control,
entering=False, special_mode=False, special_active=False):
return mqb_create_lka_hud_control(packer, bus, ldw_stock_values, enabled, steering_pressed, hud_alert, hud_control,
entering, special_mode, special_active)
def create_acc_buttons_control(packer, bus, gra_stock_values, cancel=False, resume=False, set_button=False):
@@ -32,23 +41,57 @@ def create_acc_buttons_control(packer, bus, gra_stock_values, cancel=False, resu
return packer.make_can_msg("LS_01", bus, values)
def acc_control_value(main_switch_on, long_active, cruiseOverride):
return 0
def acc_control_value(main_switch_on, long_active, cruiseOverride, accFaulted):
# ACC_01.ACC_Status_ACC uses the same enum as MQB ACC_06: 0 off, 2 standby, 3 active, 4 driver override, 6 fault
return mqb_acc_control_value(main_switch_on, long_active, cruiseOverride, accFaulted)
def acc_hud_status_value(main_switch_on, acc_faulted, longActive, longOverride):
return 0
return mqb_acc_hud_status_value(main_switch_on, acc_faulted, longActive, longOverride)
def create_acc_accel_control(packer, bus, acc_type, accel, acc_control, stopping, starting, esp_hold, comfortBand, jerkLimit):
values = {}
return [packer.make_can_msg("ACC_05", bus, values)]
def create_acc_accel_control(packer, bus, accel, acc_control, stopping):
acc_enabled = acc_control in (3, 4)
acc_01_values = {
"ACC_Status_ACC": acc_control,
"ACC_Sollbeschleunigung": accel if acc_enabled else ACC_INACTIVE_ACCEL,
"ACC_zul_Regelabw_unten": 0.2,
"ACC_zul_Regelabw_oben": 0.2,
"ACC_neg_Sollbeschl_Grad": 4.0 if acc_enabled else 0,
"ACC_pos_Sollbeschl_Grad": 4.0 if acc_enabled else 0,
"ACC_Dynamik": 3,
"ACC_Anhalten": stopping if acc_enabled else False,
"ACC_Minimale_Bremsung": 0,
}
return [packer.make_can_msg("ACC_01", bus, acc_01_values)]
def create_acc_hud_control(packer, bus, acc_hud_status, set_speed, leadDistance, distanceBars, fcw_alert, leadVisible, unavailable, decel, d_unresponsive):
values = {}
def create_acc_hud_control(packer, bus, acc_hud_status, set_speed, leadDistance, distanceBars, fcw_alert, leadVisible,
unavailable, decel, d_unresponsive, hud_text=0):
engaged = acc_hud_status in (3, 4)
priodisp = 0 if fcw_alert else 1 if (acc_hud_status == 4 or decel or leadVisible) else 2 if (acc_hud_status in (3, 2)) else 0
leadDistanceBars = distanceBars + 1 if distanceBars in (1, 2, 3) else 2
values = {
"ACC_Status_Anzeige": acc_hud_status,
"ACC_Wunschgeschw_02": set_speed if set_speed < 250 else 327.36,
"ACC_Gesetzte_Zeitluecke": leadDistanceBars,
"ACC_Anzeige_Zeitluecke": 1 if engaged else 0,
"ACC_Tachokranz": 1 if engaged else 0,
"ACC_Display_Prio": priodisp,
"ACC_Abstandsindex": leadDistance if leadVisible else 0,
"ACC_Relevantes_Objekt": 2 if fcw_alert else (1 if leadVisible else 0), # lead car: 1 green, 2 red, 0 off
"ACC_Status_Prim_Anz": 2 if fcw_alert else (1 if engaged else 0), # ACC symbol: 1 green, 2 red, 0 off
"ACC_Optischer_Fahrerhinweis": 1 if fcw_alert else 0,
"ACC_Akustik": 1 if (fcw_alert or d_unresponsive) else 0,
"ACC_Texte_Primaeranz": hud_text,
}
return packer.make_can_msg("ACC_02", bus, values)
def volkswagen_mlb_checksum(address: int, sig, d: bytearray) -> int:
xor_starting_value = {
0x109: 0x08, # ACC_01

10
iqdbc_repo/iqdbc/car/volkswagen/values.py
View File

@@ -2,7 +2,7 @@ from collections import defaultdict, namedtuple
from dataclasses import dataclass, field
from enum import Enum, IntFlag, StrEnum
from iqdbc.car import ACCELERATION_DUE_TO_GRAVITY, Bus, CanBusBase, CarSpecs, DbcDict, PlatformConfig, Platforms, structs, uds
from iqdbc.car import ACCELERATION_DUE_TO_GRAVITY, Bus, CanBusBase, CarSpecs, DbcDict, DT_CTRL, PlatformConfig, Platforms, structs, uds
from iqdbc.car.lateral import CurvatureSteeringLimits
from iqdbc.can import CANDefine
from iqdbc.car.common.conversions import Conversions as CV
@@ -204,6 +204,7 @@ class CarControllerParams:
self.STEER_DRIVER_ALLOWANCE = 60 # Driver intervention threshold 0.6 Nm
self.STEER_DELTA_UP = 9 # Max HCA reached in 0.66s (STEER_MAX / (50Hz * 0.66))
self.STEER_DELTA_DOWN = 10 # Min HCA reached in 0.60s (STEER_MAX / (50Hz * 0.60))
self.ACC_HUD_TEXT_STEP = int(2.0 / DT_CTRL) # ACC_02 primary display text dwell time
if CP.carFingerprint == CAR.PORSCHE_MACAN_MK1:
self.shifter_values = can_define.dv["Getriebe_03"]["GE_Waehlhebel"]
@@ -219,6 +220,13 @@ class CarControllerParams:
Button(structs.CarState.ButtonEvent.Type.gapAdjustCruise, "LS_01", "LS_Verstellung_Zeitluecke", [1, 2, 3]),
]
# ACC_02.ACC_Texte_Primaeranz, primary ACC display text at the bottom of the cluster
self.ACC_HUD_TEXTS = {
"none": 0,
"setSpeed": 21,
}
self.ACC_HUD_TEXT_DISTANCE = {1: 2, 2: 3, 3: 4, 4: 5} # follow distance bars to display text
else:
self.STEER_DRIVER_ALLOWANCE = 80 # Driver intervention threshold 0.8 Nm
self.STEER_DELTA_UP = 4 # Max HCA reached in 1.50s (STEER_MAX / (50Hz * 1.50))

4
iqdbc_repo/iqdbc/dbc/vw_mlb.dbc
View File

@@ -35,7 +35,7 @@ BS_:
BU_: Airbag_D4 EPB_D4 ESP_D4 Gateway_D4C7 Getriebe_AL551_951_D4_C7 Getriebe_DL501_C7 Getriebe_VL381_C7 LWS_D4 Motor_EDC17_D4 Motor_ME17_BY Motor_MED17_SIMOS8_D4 Motor_Slave_D4 QSP_D4 SAK_C7 SCR_C7 SCU_D4
BO_ 265 ACC_01: 8 Gateway_B8
SG_ ACC_01_CHK : 0|8@1+ (1,0) [0|255] "" Magnetic_Ride_LB,Motor_MLB_AU48_AU416_Diesel,Motor_MLB_B8_Q5_Otto,Motor_MLB_Q5_Hybrid
SG_ CHECKSUM : 0|8@1+ (1,0) [0|255] "" Magnetic_Ride_LB,Motor_MLB_AU48_AU416_Diesel,Motor_MLB_B8_Q5_Otto,Motor_MLB_Q5_Hybrid
SG_ COUNTER : 8|4@1+ (1,0) [0|15] "" Magnetic_Ride_LB,Motor_MLB_AU48_AU416_Diesel,Motor_MLB_B8_Q5_Otto,Motor_MLB_Q5_Hybrid
SG_ ACC_zul_Regelabw_unten : 16|6@1+ (0.024,0) [0|1.512] "Unit_MeterPerSeconSquar" Motor_MLB_AU48_AU416_Diesel,Motor_MLB_B8_Q5_Otto,Motor_MLB_Q5_Hybrid
SG_ ACC_Sollbeschleunigung : 24|11@1+ (0.005,-7.22) [-7.22|3.01] "Unit_MeterPerSeconSquar" Magnetic_Ride_LB,Motor_MLB_AU48_AU416_Diesel,Motor_MLB_B8_Q5_Otto,Motor_MLB_Q5_Hybrid
@@ -48,7 +48,7 @@ BO_ 265 ACC_01: 8 Gateway_B8
SG_ ACC_Minimale_Bremsung : 63|1@1+ (1,0) [0|1] "" Motor_MLB_AU48_AU416_Diesel,Motor_MLB_B8_Q5_Otto,Motor_MLB_Q5_Hybrid
BO_ 780 ACC_02: 8 Gateway_D4C7
SG_ ACC_02_CHK : 0|8@1+ (1,0) [0|255] "" HUD_C7,Kombi_D4
SG_ CHECKSUM : 0|8@1+ (1,0) [0|255] "" HUD_C7,Kombi_D4
SG_ COUNTER : 8|4@1+ (1,0) [0|15] "" HUD_C7,Kombi_D4
SG_ ACC_Wunschgeschw_02 : 12|10@1+ (0.32,0) [0.00|326.72] "Unit_KiloMeterPerHour" HUD_C7,Kombi_D4
SG_ ACC_Status_Prim_Anz : 22|2@1+ (1.0,0.0) [0.0|3] "" HUD_C7,Kombi_D4

2
iqdbc_repo/iqdbc/safety/modes/volkswagen_common.h
View File

@@ -100,7 +100,7 @@ void can_set_checksum(CANPacket_t *packet);
#define MSG_MOTOR_03 0x105U // RX from ECU, for driver throttle input and brake switch status
#define MSG_TSK_02 0x10CU // RX from ECU, for ACC status from drivetrain coordinator
#define MSG_ACC_05 0x10DU // RX from radar, for ACC status
#define MSG_ACC_01 0x109U // RX from radar, for ACC status (Audi B8)
#define MSG_ACC_01 0x109U // TX by OP, ACC control instructions to the drivetrain coordinator
static void volkswagen_common_init(void) {
volkswagen_set_button_prev = false;

50
iqdbc_repo/iqdbc/safety/modes/volkswagen_mlb.h
View File

@@ -9,6 +9,9 @@ static safety_config volkswagen_mlb_init(uint16_t param) {
static const CanMsg VOLKSWAGEN_MLB_STOCK_TX_MSGS[] = {{MSG_HCA_01, 0, 8, .check_relay = true}, {MSG_LDW_02, 0, 8, .check_relay = true},
{MSG_LS_01, 0, 4, .check_relay = false}, {MSG_LS_01, 2, 4, .check_relay = false}};
static const CanMsg VOLKSWAGEN_MLB_LONG_TX_MSGS[] = {{MSG_HCA_01, 0, 8, .check_relay = true}, {MSG_LDW_02, 0, 8, .check_relay = true},
{MSG_ACC_01, 0, 8, .check_relay = true}, {MSG_ACC_02, 0, 8, .check_relay = true}};
static RxCheck volkswagen_mlb_rx_checks[] = {
// TODO: implement checksum validation
{.msg = {{MSG_ESP_03, 0, 8, 50U, .ignore_checksum = true, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}},
@@ -19,10 +22,17 @@ static safety_config volkswagen_mlb_init(uint16_t param) {
{.msg = {{MSG_LS_01, 0, 4, 10U, .ignore_checksum = true, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}},
};
SAFETY_UNUSED(param);
volkswagen_common_init();
return BUILD_SAFETY_CFG(volkswagen_mlb_rx_checks, VOLKSWAGEN_MLB_STOCK_TX_MSGS);
#ifdef ALLOW_DEBUG
volkswagen_longitudinal = GET_FLAG(param, FLAG_VOLKSWAGEN_LONG_CONTROL);
volkswagen_allow_long_accel_with_gas_pressed = GET_FLAG(param, FLAG_VOLKSWAGEN_ALLOW_LONG_ACCEL_WITH_GAS_PRESSED);
#else
SAFETY_UNUSED(param);
#endif
return volkswagen_longitudinal ? BUILD_SAFETY_CFG(volkswagen_mlb_rx_checks, VOLKSWAGEN_MLB_LONG_TX_MSGS) : \
BUILD_SAFETY_CFG(volkswagen_mlb_rx_checks, VOLKSWAGEN_MLB_STOCK_TX_MSGS);
}
static void volkswagen_mlb_rx_hook(const CANPacket_t *msg) {
@@ -44,6 +54,27 @@ static void volkswagen_mlb_rx_hook(const CANPacket_t *msg) {
}
if (msg->addr == MSG_LS_01) {
// If using openpilot longitudinal, the stock ACC coordinator is relayed out, so the stalk main
// switch is the only remaining source of truth. Enter controls on falling edge of Set or Resume.
// Signal: LS_01.LS_Hauptschalter
// Signal: LS_01.LS_Tip_Setzen
// Signal: LS_01.LS_Tip_Wiederaufnahme
if (volkswagen_longitudinal) {
acc_main_on = GET_BIT(msg, 12U);
bool set_button = GET_BIT(msg, 16U);
bool resume_button = GET_BIT(msg, 19U);
if ((volkswagen_set_button_prev && !set_button) || (volkswagen_resume_button_prev && !resume_button)) {
controls_allowed = acc_main_on;
}
volkswagen_set_button_prev = set_button;
volkswagen_resume_button_prev = resume_button;
if (!acc_main_on) {
controls_allowed = false;
}
}
// Always exit controls on rising edge of Cancel
// Signal: LS_01.LS_Abbrechen
if (GET_BIT(msg, 13U)) {
@@ -64,7 +95,7 @@ static void volkswagen_mlb_rx_hook(const CANPacket_t *msg) {
brake_pressed = volkswagen_brake_pedal_switch || volkswagen_brake_pressure_detected;
if (msg->addr == MSG_TSK_02) {
if ((msg->addr == MSG_TSK_02) && !volkswagen_longitudinal) {
// When using stock ACC, enter controls on rising edge of stock ACC engage, exit on disengage
// Always exit controls on main switch off
// Signal: TSK_02.TSK_Status
@@ -80,7 +111,7 @@ static void volkswagen_mlb_rx_hook(const CANPacket_t *msg) {
if (msg->bus == 2U) {
// TODO: See if there's a bus-agnostic TSK message we can use instead
if (msg->addr == MSG_ACC_05) {
if ((msg->addr == MSG_ACC_05) && !volkswagen_longitudinal) {
// When using stock ACC, enter controls on rising edge of stock ACC engage, exit on disengage
// Always exit controls on main switch off
// Signal: ACC_05.ACC_Status_ACC
@@ -123,6 +154,17 @@ static bool volkswagen_mlb_tx_hook(const CANPacket_t *msg) {
}
}
// Safety check for ACC_01 acceleration request
// Signal: ACC_01.ACC_Sollbeschleunigung (acceleration in m/s^2, scale 0.005, offset -7.22)
// To avoid floating point math, scale upward and compare to pre-scaled safety m/s^2 boundaries
if (msg->addr == MSG_ACC_01) {
int desired_accel = ((((msg->data[4] & 0x07U) << 8) | msg->data[3]) * 5U) - 7220U;
if (volkswagen_iq_long_accel_check(desired_accel)) {
tx = false;
}
}
// FORCE CANCEL: ensuring that only the cancel button press is sent when controls are off.
// This avoids unintended engagements while still allowing resume spam
if ((msg->addr == MSG_LS_01) && !controls_allowed) {

8
iqdbc_repo/iqdbc/safety/tests/common.py
View File

@@ -926,12 +926,12 @@ class SafetyTest(SafetyTestBase):
if attr.startswith('TestSubaru') and current_test == 'TestVolkswagenMqbLongSafety':
tx = list(filter(lambda m: m[0] not in [0x122, ], tx))
# Volkswagen MQB and Honda Nidec ACC HUD messages overlap
if attr == 'TestVolkswagenMqbLongSafety' and current_test.startswith('TestHondaNidec'):
# Volkswagen MQB/MLB and Honda Nidec ACC HUD messages overlap
if attr in ('TestVolkswagenMqbLongSafety', 'TestVolkswagenMlbLongSafety') and current_test.startswith('TestHondaNidec'):
tx = list(filter(lambda m: m[0] not in [0x30c, ], tx))
# Volkswagen MQB and Honda Bosch Radarless ACC HUD messages overlap
if attr == 'TestVolkswagenMqbLongSafety' and current_test.startswith('TestHondaBoschRadarless'):
# Volkswagen MQB/MLB and Honda Bosch Radarless ACC HUD messages overlap
if attr in ('TestVolkswagenMqbLongSafety', 'TestVolkswagenMlbLongSafety') and current_test.startswith('TestHondaBoschRadarless'):
tx = list(filter(lambda m: m[0] not in [0x30c, ], tx))
# TODO: Temporary, should be fixed in panda firmware, safety_honda.h

75
iqdbc_repo/iqdbc/safety/tests/test_volkswagen_mlb.py
View File

@@ -1,17 +1,24 @@
#!/usr/bin/env python3
import unittest
import numpy as np
from iqdbc.car.structs import CarParams
from iqdbc.safety.tests.libsafety import libsafety_py
import iqdbc.safety.tests.common as common
from iqdbc.safety.tests.common import CANPackerSafety
from iqdbc.car.volkswagen.values import VolkswagenSafetyFlags
MAX_ACCEL = 2.0
MIN_ACCEL = -3.5
MSG_LH_EPS_03 = 0x9F # RX from EPS, for driver steering torque
MSG_ACC_01 = 0x109 # TX by OP, ACC acceleration request to the drivetrain coordinator
MSG_ESP_03 = 0x103 # RX from ABS, for wheel speeds
MSG_MOTOR_03 = 0x105 # RX from ECU, for driver throttle input and driver brake input
MSG_ESP_05 = 0x106 # RX from ABS, for brake light state
MSG_LS_01 = 0x10B # TX by OP, ACC control buttons for cancel/resume
MSG_TSK_02 = 0x10C # RX from ECU, for ACC status from drivetrain coordinator
MSG_HCA_01 = 0x126 # TX by OP, Heading Control Assist steering torque
MSG_ACC_02 = 0x30C # TX by OP, ACC HUD data to the instrument cluster
MSG_LDW_02 = 0x397 # TX by OP, Lane line recognition and text alerts
@@ -72,10 +79,16 @@ class TestVolkswagenMlbSafetyBase(common.CarSafetyTest, common.DriverTorqueSteer
return self.packer.make_can_msg_safety("HCA_01", 0, values)
# Cruise control buttons
def _ls_01_msg(self, cancel=0, resume=0, _set=0, bus=2):
values = {"LS_Abbrechen": cancel, "LS_Tip_Setzen": _set, "LS_Tip_Wiederaufnahme": resume}
def _ls_01_msg(self, cancel=0, resume=0, _set=0, main_switch=1, bus=2):
values = {"LS_Abbrechen": cancel, "LS_Tip_Setzen": _set, "LS_Tip_Wiederaufnahme": resume,
"LS_Hauptschalter": main_switch}
return self.packer.make_can_msg_safety("LS_01", bus, values)
# Acceleration request to drivetrain coordinator
def _acc_01_msg(self, accel):
values = {"ACC_Sollbeschleunigung": accel}
return self.packer.make_can_msg_safety("ACC_01", 0, values)
# Verify brake_pressed is true if either the switch or pressure threshold signals are true
def test_redundant_brake_signals(self):
test_combinations = [(True, True, True), (True, True, False), (True, False, True), (False, False, False)]
@@ -137,5 +150,63 @@ class TestVolkswagenMlbStockSafety(TestVolkswagenMlbSafetyBase):
self.assertFalse(self.safety.get_controls_allowed(), "controls allowed after cancel")
class TestVolkswagenMlbLongSafety(TestVolkswagenMlbSafetyBase):
TX_MSGS = [[MSG_HCA_01, 0], [MSG_LDW_02, 0], [MSG_ACC_01, 0], [MSG_ACC_02, 0]]
FWD_BLACKLISTED_ADDRS = {2: [MSG_HCA_01, MSG_LDW_02, MSG_ACC_01, MSG_ACC_02]}
FWD_BUS_LOOKUP = {0: 2, 2: 0}
RELAY_MALFUNCTION_ADDRS = {0: (MSG_HCA_01, MSG_LDW_02, MSG_ACC_01, MSG_ACC_02)}
INACTIVE_ACCEL = 3.01
def setUp(self):
self.packer = CANPackerSafety("vw_mlb")
self.safety = libsafety_py.libsafety
safety_param = VolkswagenSafetyFlags.LONG_CONTROL | VolkswagenSafetyFlags.ALLOW_LONG_ACCEL_WITH_GAS_PRESSED
self.safety.set_safety_hooks(CarParams.SafetyModel.volkswagenMlb, safety_param)
self.safety.init_tests()
# stock cruise controls are entirely bypassed under openpilot longitudinal control
def test_disable_control_allowed_from_cruise(self):
pass
def test_enable_control_allowed_from_cruise(self):
pass
def test_cruise_engaged_prev(self):
pass
def test_set_and_resume_buttons(self):
for button in ["set", "resume"]:
# ACC main switch must be on, engage on falling edge
self.safety.set_controls_allowed(0)
self._rx(self._ls_01_msg(_set=(button == "set"), resume=(button == "resume"), main_switch=0, bus=0))
self.assertFalse(self.safety.get_controls_allowed(), f"controls allowed on {button} with main switch off")
self._rx(self._ls_01_msg(main_switch=0, bus=0))
self._rx(self._ls_01_msg(_set=(button == "set"), resume=(button == "resume"), bus=0))
self.assertFalse(self.safety.get_controls_allowed(), f"controls allowed on {button} rising edge")
self._rx(self._ls_01_msg(bus=0))
self.assertTrue(self.safety.get_controls_allowed(), f"controls not allowed on {button} falling edge")
def test_main_switch(self):
# Disable as soon as the ACC main switch turns off
self._rx(self._ls_01_msg(bus=0))
self.safety.set_controls_allowed(1)
self._rx(self._ls_01_msg(main_switch=0, bus=0))
self.assertFalse(self.safety.get_controls_allowed(), "controls allowed after ACC main switch off")
def test_accel_safety_check(self):
for controls_allowed in [True, False]:
for accel in np.concatenate((np.arange(MIN_ACCEL - 2, MAX_ACCEL + 2, 0.03), [0, self.INACTIVE_ACCEL])):
accel = round(accel, 2)
is_inactive_accel = accel == self.INACTIVE_ACCEL
send = (controls_allowed and MIN_ACCEL <= accel <= MAX_ACCEL) or is_inactive_accel
self.safety.set_controls_allowed(controls_allowed)
self.assertEqual(send, self._tx(self._acc_01_msg(accel)), (controls_allowed, accel))
def test_accel_allowed_with_gas_pressed(self):
self._rx(self._user_gas_msg(1))
self.safety.set_controls_allowed(True)
self.assertTrue(self._tx(self._acc_01_msg(0.5)))
if __name__ == "__main__":
unittest.main()