clearpilot: initial commit of full source

This commit is contained in:
2026-04-11 06:25:25 +00:00
commit e2a0c1894a
3383 changed files with 834683 additions and 0 deletions

View File

View File

@@ -0,0 +1,123 @@
from cereal import car
from opendbc.can.packer import CANPacker
from openpilot.common.numpy_fast import clip
from openpilot.common.conversions import Conversions as CV
from openpilot.common.realtime import DT_CTRL
from openpilot.selfdrive.car import apply_driver_steer_torque_limits
from openpilot.selfdrive.car.interfaces import CarControllerBase
from openpilot.selfdrive.car.volkswagen import mqbcan, pqcan
from openpilot.selfdrive.car.volkswagen.values import CANBUS, CarControllerParams, VolkswagenFlags
VisualAlert = car.CarControl.HUDControl.VisualAlert
LongCtrlState = car.CarControl.Actuators.LongControlState
class CarController(CarControllerBase):
def __init__(self, dbc_name, CP, VM):
self.CP = CP
self.CCP = CarControllerParams(CP)
self.CCS = pqcan if CP.flags & VolkswagenFlags.PQ else mqbcan
self.packer_pt = CANPacker(dbc_name)
self.apply_steer_last = 0
self.gra_acc_counter_last = None
self.frame = 0
self.eps_timer_soft_disable_alert = False
self.hca_frame_timer_running = 0
self.hca_frame_same_torque = 0
def update(self, CC, CS, ext_bus, now_nanos, frogpilot_variables):
actuators = CC.actuators
hud_control = CC.hudControl
can_sends = []
# **** Steering Controls ************************************************ #
if self.frame % self.CCP.STEER_STEP == 0:
# Logic to avoid HCA state 4 "refused":
# * Don't steer unless HCA is in state 3 "ready" or 5 "active"
# * Don't steer at standstill
# * Don't send > 3.00 Newton-meters torque
# * Don't send the same torque for > 6 seconds
# * Don't send uninterrupted steering for > 360 seconds
# MQB racks reset the uninterrupted steering timer after a single frame
# of HCA disabled; this is done whenever output happens to be zero.
if CC.latActive:
new_steer = int(round(actuators.steer * self.CCP.STEER_MAX))
apply_steer = apply_driver_steer_torque_limits(new_steer, self.apply_steer_last, CS.out.steeringTorque, self.CCP)
self.hca_frame_timer_running += self.CCP.STEER_STEP
if self.apply_steer_last == apply_steer:
self.hca_frame_same_torque += self.CCP.STEER_STEP
if self.hca_frame_same_torque > self.CCP.STEER_TIME_STUCK_TORQUE / DT_CTRL:
apply_steer -= (1, -1)[apply_steer < 0]
self.hca_frame_same_torque = 0
else:
self.hca_frame_same_torque = 0
hca_enabled = abs(apply_steer) > 0
else:
hca_enabled = False
apply_steer = 0
if not hca_enabled:
self.hca_frame_timer_running = 0
self.eps_timer_soft_disable_alert = self.hca_frame_timer_running > self.CCP.STEER_TIME_ALERT / DT_CTRL
self.apply_steer_last = apply_steer
can_sends.append(self.CCS.create_steering_control(self.packer_pt, CANBUS.pt, apply_steer, hca_enabled))
if self.CP.flags & VolkswagenFlags.STOCK_HCA_PRESENT:
# Pacify VW Emergency Assist driver inactivity detection by changing its view of driver steering input torque
# to the greatest of actual driver input or 2x openpilot's output (1x openpilot output is not enough to
# consistently reset inactivity detection on straight level roads). See commaai/openpilot#23274 for background.
ea_simulated_torque = clip(apply_steer * 2, -self.CCP.STEER_MAX, self.CCP.STEER_MAX)
if abs(CS.out.steeringTorque) > abs(ea_simulated_torque):
ea_simulated_torque = CS.out.steeringTorque
can_sends.append(self.CCS.create_eps_update(self.packer_pt, CANBUS.cam, CS.eps_stock_values, ea_simulated_torque))
# **** Acceleration Controls ******************************************** #
if self.frame % self.CCP.ACC_CONTROL_STEP == 0 and self.CP.openpilotLongitudinalControl:
acc_control = self.CCS.acc_control_value(CS.out.cruiseState.available, CS.out.accFaulted, CC.longActive)
if frogpilot_variables.sport_plus:
accel = clip(actuators.accel, self.CCP.ACCEL_MIN, self.CCP.ACCEL_MAX_PLUS) if CC.longActive else 0
else:
accel = clip(actuators.accel, self.CCP.ACCEL_MIN, self.CCP.ACCEL_MAX) if CC.longActive else 0
stopping = actuators.longControlState == LongCtrlState.stopping
starting = actuators.longControlState == LongCtrlState.pid and (CS.esp_hold_confirmation or CS.out.vEgo < self.CP.vEgoStopping)
can_sends.extend(self.CCS.create_acc_accel_control(self.packer_pt, CANBUS.pt, CS.acc_type, CC.longActive, accel,
acc_control, stopping, starting, CS.esp_hold_confirmation))
# **** HUD Controls ***************************************************** #
if self.frame % self.CCP.LDW_STEP == 0:
hud_alert = 0
if hud_control.visualAlert in (VisualAlert.steerRequired, VisualAlert.ldw):
hud_alert = self.CCP.LDW_MESSAGES["laneAssistTakeOver"]
can_sends.append(self.CCS.create_lka_hud_control(self.packer_pt, CANBUS.pt, CS.ldw_stock_values, CC.enabled,
CS.out.steeringPressed, hud_alert, hud_control, CC.latActive))
if self.frame % self.CCP.ACC_HUD_STEP == 0 and self.CP.openpilotLongitudinalControl:
lead_distance = 0
if hud_control.leadVisible and self.frame * DT_CTRL > 1.0: # Don't display lead until we know the scaling factor
lead_distance = 512 if CS.upscale_lead_car_signal else 8
acc_hud_status = self.CCS.acc_hud_status_value(CS.out.cruiseState.available, CS.out.accFaulted, CC.longActive)
# FIXME: follow the recent displayed-speed updates, also use mph_kmh toggle to fix display rounding problem?
set_speed = hud_control.setSpeed * CV.MS_TO_KPH
can_sends.append(self.CCS.create_acc_hud_control(self.packer_pt, CANBUS.pt, acc_hud_status, set_speed,
lead_distance, hud_control.leadDistanceBars))
# **** Stock ACC Button Controls **************************************** #
gra_send_ready = self.CP.pcmCruise and CS.gra_stock_values["COUNTER"] != self.gra_acc_counter_last
if gra_send_ready and (CC.cruiseControl.cancel or CC.cruiseControl.resume):
can_sends.append(self.CCS.create_acc_buttons_control(self.packer_pt, ext_bus, CS.gra_stock_values,
cancel=CC.cruiseControl.cancel, resume=CC.cruiseControl.resume))
new_actuators = actuators.copy()
new_actuators.steer = self.apply_steer_last / self.CCP.STEER_MAX
new_actuators.steerOutputCan = self.apply_steer_last
self.gra_acc_counter_last = CS.gra_stock_values["COUNTER"]
self.frame += 1
return new_actuators, can_sends, self.eps_timer_soft_disable_alert

View File

@@ -0,0 +1,395 @@
import numpy as np
from cereal import car
from openpilot.common.conversions import Conversions as CV
from openpilot.selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from openpilot.selfdrive.car.volkswagen.values import DBC, CANBUS, NetworkLocation, TransmissionType, GearShifter, \
CarControllerParams, VolkswagenFlags
class CarState(CarStateBase):
def __init__(self, CP):
super().__init__(CP)
self.frame = 0
self.eps_init_complete = False
self.CCP = CarControllerParams(CP)
self.button_states = {button.event_type: False for button in self.CCP.BUTTONS}
self.esp_hold_confirmation = False
self.upscale_lead_car_signal = False
self.eps_stock_values = False
def create_button_events(self, pt_cp, buttons):
button_events = []
for button in buttons:
state = pt_cp.vl[button.can_addr][button.can_msg] in button.values
if self.button_states[button.event_type] != state:
event = car.CarState.ButtonEvent.new_message()
event.type = button.event_type
event.pressed = state
button_events.append(event)
self.button_states[button.event_type] = state
return button_events
def update(self, pt_cp, cam_cp, ext_cp, trans_type, frogpilot_variables):
if self.CP.flags & VolkswagenFlags.PQ:
return self.update_pq(pt_cp, cam_cp, ext_cp, trans_type, frogpilot_variables)
ret = car.CarState.new_message()
# Update vehicle speed and acceleration from ABS wheel speeds.
ret.wheelSpeeds = self.get_wheel_speeds(
pt_cp.vl["ESP_19"]["ESP_VL_Radgeschw_02"],
pt_cp.vl["ESP_19"]["ESP_VR_Radgeschw_02"],
pt_cp.vl["ESP_19"]["ESP_HL_Radgeschw_02"],
pt_cp.vl["ESP_19"]["ESP_HR_Radgeschw_02"],
)
ret.vEgoRaw = float(np.mean([ret.wheelSpeeds.fl, ret.wheelSpeeds.fr, ret.wheelSpeeds.rl, ret.wheelSpeeds.rr]))
ret.vEgo, ret.aEgo = self.update_speed_kf(ret.vEgoRaw)
ret.standstill = ret.vEgoRaw == 0
# Update EPS position and state info. For signed values, VW sends the sign in a separate signal.
ret.steeringAngleDeg = pt_cp.vl["LWI_01"]["LWI_Lenkradwinkel"] * (1, -1)[int(pt_cp.vl["LWI_01"]["LWI_VZ_Lenkradwinkel"])]
ret.steeringRateDeg = pt_cp.vl["LWI_01"]["LWI_Lenkradw_Geschw"] * (1, -1)[int(pt_cp.vl["LWI_01"]["LWI_VZ_Lenkradw_Geschw"])]
ret.steeringTorque = pt_cp.vl["LH_EPS_03"]["EPS_Lenkmoment"] * (1, -1)[int(pt_cp.vl["LH_EPS_03"]["EPS_VZ_Lenkmoment"])]
ret.steeringPressed = abs(ret.steeringTorque) > self.CCP.STEER_DRIVER_ALLOWANCE
ret.yawRate = pt_cp.vl["ESP_02"]["ESP_Gierrate"] * (1, -1)[int(pt_cp.vl["ESP_02"]["ESP_VZ_Gierrate"])] * CV.DEG_TO_RAD
hca_status = self.CCP.hca_status_values.get(pt_cp.vl["LH_EPS_03"]["EPS_HCA_Status"])
ret.steerFaultTemporary, ret.steerFaultPermanent = self.update_hca_state(hca_status)
# VW Emergency Assist status tracking and mitigation
self.eps_stock_values = pt_cp.vl["LH_EPS_03"]
if self.CP.flags & VolkswagenFlags.STOCK_HCA_PRESENT:
ret.carFaultedNonCritical = bool(cam_cp.vl["HCA_01"]["EA_Ruckfreigabe"]) or cam_cp.vl["HCA_01"]["EA_ACC_Sollstatus"] > 0
# Update gas, brakes, and gearshift.
ret.gas = pt_cp.vl["Motor_20"]["MO_Fahrpedalrohwert_01"] / 100.0
ret.gasPressed = ret.gas > 0
ret.brake = pt_cp.vl["ESP_05"]["ESP_Bremsdruck"] / 250.0 # FIXME: this is pressure in Bar, not sure what OP expects
brake_pedal_pressed = bool(pt_cp.vl["Motor_14"]["MO_Fahrer_bremst"])
brake_pressure_detected = bool(pt_cp.vl["ESP_05"]["ESP_Fahrer_bremst"])
ret.brakePressed = brake_pedal_pressed or brake_pressure_detected
ret.parkingBrake = bool(pt_cp.vl["Kombi_01"]["KBI_Handbremse"]) # FIXME: need to include an EPB check as well
# Update gear and/or clutch position data.
if trans_type == TransmissionType.automatic:
ret.gearShifter = self.parse_gear_shifter(self.CCP.shifter_values.get(pt_cp.vl["Getriebe_11"]["GE_Fahrstufe"], None))
elif trans_type == TransmissionType.direct:
ret.gearShifter = self.parse_gear_shifter(self.CCP.shifter_values.get(pt_cp.vl["EV_Gearshift"]["GearPosition"], None))
elif trans_type == TransmissionType.manual:
ret.clutchPressed = not pt_cp.vl["Motor_14"]["MO_Kuppl_schalter"]
if bool(pt_cp.vl["Gateway_72"]["BCM1_Rueckfahrlicht_Schalter"]):
ret.gearShifter = GearShifter.reverse
else:
ret.gearShifter = GearShifter.drive
# Update door and trunk/hatch lid open status.
ret.doorOpen = any([pt_cp.vl["Gateway_72"]["ZV_FT_offen"],
pt_cp.vl["Gateway_72"]["ZV_BT_offen"],
pt_cp.vl["Gateway_72"]["ZV_HFS_offen"],
pt_cp.vl["Gateway_72"]["ZV_HBFS_offen"],
pt_cp.vl["Gateway_72"]["ZV_HD_offen"]])
# Update seatbelt fastened status.
ret.seatbeltUnlatched = pt_cp.vl["Airbag_02"]["AB_Gurtschloss_FA"] != 3
# Consume blind-spot monitoring info/warning LED states, if available.
# Infostufe: BSM LED on, Warnung: BSM LED flashing
if self.CP.enableBsm:
ret.leftBlindspot = bool(ext_cp.vl["SWA_01"]["SWA_Infostufe_SWA_li"]) or bool(ext_cp.vl["SWA_01"]["SWA_Warnung_SWA_li"])
ret.rightBlindspot = bool(ext_cp.vl["SWA_01"]["SWA_Infostufe_SWA_re"]) or bool(ext_cp.vl["SWA_01"]["SWA_Warnung_SWA_re"])
# Consume factory LDW data relevant for factory SWA (Lane Change Assist)
# and capture it for forwarding to the blind spot radar controller
self.ldw_stock_values = cam_cp.vl["LDW_02"] if self.CP.networkLocation == NetworkLocation.fwdCamera else {}
# Stock FCW is considered active if the release bit for brake-jerk warning
# is set. Stock AEB considered active if the partial braking or target
# braking release bits are set.
# Refer to VW Self Study Program 890253: Volkswagen Driver Assistance
# Systems, chapter on Front Assist with Braking: Golf Family for all MQB
ret.stockFcw = bool(ext_cp.vl["ACC_10"]["AWV2_Freigabe"])
ret.stockAeb = bool(ext_cp.vl["ACC_10"]["ANB_Teilbremsung_Freigabe"]) or bool(ext_cp.vl["ACC_10"]["ANB_Zielbremsung_Freigabe"])
# Update ACC radar status.
self.acc_type = ext_cp.vl["ACC_06"]["ACC_Typ"]
if pt_cp.vl["TSK_06"]["TSK_Status"] == 2:
# ACC okay and enabled, but not currently engaged
ret.cruiseState.available = True
ret.cruiseState.enabled = False
elif pt_cp.vl["TSK_06"]["TSK_Status"] in (3, 4, 5):
# ACC okay and enabled, currently regulating speed (3) or driver accel override (4) or brake only (5)
ret.cruiseState.available = True
ret.cruiseState.enabled = True
else:
# ACC okay but disabled (1), or a radar visibility or other fault/disruption (6 or 7)
ret.cruiseState.available = False
ret.cruiseState.enabled = False
self.esp_hold_confirmation = bool(pt_cp.vl["ESP_21"]["ESP_Haltebestaetigung"])
ret.cruiseState.standstill = self.CP.pcmCruise and self.esp_hold_confirmation
ret.accFaulted = pt_cp.vl["TSK_06"]["TSK_Status"] in (6, 7)
# Update ACC setpoint. When the setpoint is zero or there's an error, the
# radar sends a set-speed of ~90.69 m/s / 203mph.
if self.CP.pcmCruise:
ret.cruiseState.speed = ext_cp.vl["ACC_02"]["ACC_Wunschgeschw_02"] * CV.KPH_TO_MS
if ret.cruiseState.speed > 90:
ret.cruiseState.speed = 0
# Update button states for turn signals and ACC controls, capture all ACC button state/config for passthrough
ret.leftBlinker = bool(pt_cp.vl["Blinkmodi_02"]["Comfort_Signal_Left"])
ret.rightBlinker = bool(pt_cp.vl["Blinkmodi_02"]["Comfort_Signal_Right"])
ret.buttonEvents = self.create_button_events(pt_cp, self.CCP.BUTTONS)
self.gra_stock_values = pt_cp.vl["GRA_ACC_01"]
# Additional safety checks performed in CarInterface.
ret.espDisabled = pt_cp.vl["ESP_21"]["ESP_Tastung_passiv"] != 0
# Digital instrument clusters expect the ACC HUD lead car distance to be scaled differently
self.upscale_lead_car_signal = bool(pt_cp.vl["Kombi_03"]["KBI_Variante"])
self.frame += 1
return ret
def update_pq(self, pt_cp, cam_cp, ext_cp, trans_type, frogpilot_variables):
ret = car.CarState.new_message()
# Update vehicle speed and acceleration from ABS wheel speeds.
ret.wheelSpeeds = self.get_wheel_speeds(
pt_cp.vl["Bremse_3"]["Radgeschw__VL_4_1"],
pt_cp.vl["Bremse_3"]["Radgeschw__VR_4_1"],
pt_cp.vl["Bremse_3"]["Radgeschw__HL_4_1"],
pt_cp.vl["Bremse_3"]["Radgeschw__HR_4_1"],
)
# vEgo obtained from Bremse_1 vehicle speed rather than Bremse_3 wheel speeds because Bremse_3 isn't present on NSF
ret.vEgoRaw = pt_cp.vl["Bremse_1"]["Geschwindigkeit_neu__Bremse_1_"] * CV.KPH_TO_MS
ret.vEgo, ret.aEgo = self.update_speed_kf(ret.vEgoRaw)
ret.standstill = ret.vEgoRaw == 0
# Update EPS position and state info. For signed values, VW sends the sign in a separate signal.
ret.steeringAngleDeg = pt_cp.vl["Lenkhilfe_3"]["LH3_BLW"] * (1, -1)[int(pt_cp.vl["Lenkhilfe_3"]["LH3_BLWSign"])]
ret.steeringRateDeg = pt_cp.vl["Lenkwinkel_1"]["Lenkradwinkel_Geschwindigkeit"] * (1, -1)[int(pt_cp.vl["Lenkwinkel_1"]["Lenkradwinkel_Geschwindigkeit_S"])]
ret.steeringTorque = pt_cp.vl["Lenkhilfe_3"]["LH3_LM"] * (1, -1)[int(pt_cp.vl["Lenkhilfe_3"]["LH3_LMSign"])]
ret.steeringPressed = abs(ret.steeringTorque) > self.CCP.STEER_DRIVER_ALLOWANCE
ret.yawRate = pt_cp.vl["Bremse_5"]["Giergeschwindigkeit"] * (1, -1)[int(pt_cp.vl["Bremse_5"]["Vorzeichen_der_Giergeschwindigk"])] * CV.DEG_TO_RAD
hca_status = self.CCP.hca_status_values.get(pt_cp.vl["Lenkhilfe_2"]["LH2_Sta_HCA"])
ret.steerFaultTemporary, ret.steerFaultPermanent = self.update_hca_state(hca_status)
# Update gas, brakes, and gearshift.
ret.gas = pt_cp.vl["Motor_3"]["Fahrpedal_Rohsignal"] / 100.0
ret.gasPressed = ret.gas > 0
ret.brake = pt_cp.vl["Bremse_5"]["Bremsdruck"] / 250.0 # FIXME: this is pressure in Bar, not sure what OP expects
ret.brakePressed = bool(pt_cp.vl["Motor_2"]["Bremslichtschalter"])
ret.parkingBrake = bool(pt_cp.vl["Kombi_1"]["Bremsinfo"])
# Update gear and/or clutch position data.
if trans_type == TransmissionType.automatic:
ret.gearShifter = self.parse_gear_shifter(self.CCP.shifter_values.get(pt_cp.vl["Getriebe_1"]["Waehlhebelposition__Getriebe_1_"], None))
elif trans_type == TransmissionType.manual:
ret.clutchPressed = not pt_cp.vl["Motor_1"]["Kupplungsschalter"]
reverse_light = bool(pt_cp.vl["Gate_Komf_1"]["GK1_Rueckfahr"])
if reverse_light:
ret.gearShifter = GearShifter.reverse
else:
ret.gearShifter = GearShifter.drive
# Update door and trunk/hatch lid open status.
ret.doorOpen = any([pt_cp.vl["Gate_Komf_1"]["GK1_Fa_Tuerkont"],
pt_cp.vl["Gate_Komf_1"]["BSK_BT_geoeffnet"],
pt_cp.vl["Gate_Komf_1"]["BSK_HL_geoeffnet"],
pt_cp.vl["Gate_Komf_1"]["BSK_HR_geoeffnet"],
pt_cp.vl["Gate_Komf_1"]["BSK_HD_Hauptraste"]])
# Update seatbelt fastened status.
ret.seatbeltUnlatched = not bool(pt_cp.vl["Airbag_1"]["Gurtschalter_Fahrer"])
# Consume blind-spot monitoring info/warning LED states, if available.
# Infostufe: BSM LED on, Warnung: BSM LED flashing
if self.CP.enableBsm:
ret.leftBlindspot = bool(ext_cp.vl["SWA_1"]["SWA_Infostufe_SWA_li"]) or bool(ext_cp.vl["SWA_1"]["SWA_Warnung_SWA_li"])
ret.rightBlindspot = bool(ext_cp.vl["SWA_1"]["SWA_Infostufe_SWA_re"]) or bool(ext_cp.vl["SWA_1"]["SWA_Warnung_SWA_re"])
# Consume factory LDW data relevant for factory SWA (Lane Change Assist)
# and capture it for forwarding to the blind spot radar controller
self.ldw_stock_values = cam_cp.vl["LDW_Status"] if self.CP.networkLocation == NetworkLocation.fwdCamera else {}
# Stock FCW is considered active if the release bit for brake-jerk warning
# is set. Stock AEB considered active if the partial braking or target
# braking release bits are set.
# Refer to VW Self Study Program 890253: Volkswagen Driver Assistance
# Systems, chapters on Front Assist with Braking and City Emergency
# Braking for the 2016 Passat NMS
# TODO: deferred until we can collect data on pre-MY2016 behavior, AWV message may be shorter with fewer signals
ret.stockFcw = False
ret.stockAeb = False
# Update ACC radar status.
self.acc_type = ext_cp.vl["ACC_System"]["ACS_Typ_ACC"]
ret.cruiseState.available = bool(pt_cp.vl["Motor_5"]["GRA_Hauptschalter"])
ret.cruiseState.enabled = pt_cp.vl["Motor_2"]["GRA_Status"] in (1, 2)
if self.CP.pcmCruise:
ret.accFaulted = ext_cp.vl["ACC_GRA_Anzeige"]["ACA_StaACC"] in (6, 7)
else:
ret.accFaulted = pt_cp.vl["Motor_2"]["GRA_Status"] == 3
# Update ACC setpoint. When the setpoint reads as 255, the driver has not
# yet established an ACC setpoint, so treat it as zero.
ret.cruiseState.speed = ext_cp.vl["ACC_GRA_Anzeige"]["ACA_V_Wunsch"] * CV.KPH_TO_MS
if ret.cruiseState.speed > 70: # 255 kph in m/s == no current setpoint
ret.cruiseState.speed = 0
# Update button states for turn signals and ACC controls, capture all ACC button state/config for passthrough
ret.leftBlinker, ret.rightBlinker = self.update_blinker_from_stalk(300, pt_cp.vl["Gate_Komf_1"]["GK1_Blinker_li"],
pt_cp.vl["Gate_Komf_1"]["GK1_Blinker_re"])
ret.buttonEvents = self.create_button_events(pt_cp, self.CCP.BUTTONS)
self.gra_stock_values = pt_cp.vl["GRA_Neu"]
# Additional safety checks performed in CarInterface.
ret.espDisabled = bool(pt_cp.vl["Bremse_1"]["ESP_Passiv_getastet"])
self.frame += 1
return ret
def update_hca_state(self, hca_status):
# Treat INITIALIZING and FAULT as temporary for worst likely EPS recovery time, for cars without factory Lane Assist
# DISABLED means the EPS hasn't been configured to support Lane Assist
self.eps_init_complete = self.eps_init_complete or (hca_status in ("DISABLED", "READY", "ACTIVE") or self.frame > 600)
perm_fault = hca_status == "DISABLED" or (self.eps_init_complete and hca_status in ("INITIALIZING", "FAULT"))
temp_fault = hca_status == "REJECTED" or not self.eps_init_complete
return temp_fault, perm_fault
@staticmethod
def get_can_parser(CP):
if CP.flags & VolkswagenFlags.PQ:
return CarState.get_can_parser_pq(CP)
messages = [
# sig_address, frequency
("LWI_01", 100), # From J500 Steering Assist with integrated sensors
("LH_EPS_03", 100), # From J500 Steering Assist with integrated sensors
("ESP_19", 100), # From J104 ABS/ESP controller
("ESP_05", 50), # From J104 ABS/ESP controller
("ESP_21", 50), # From J104 ABS/ESP controller
("Motor_20", 50), # From J623 Engine control module
("TSK_06", 50), # From J623 Engine control module
("ESP_02", 50), # From J104 ABS/ESP controller
("GRA_ACC_01", 33), # From J533 CAN gateway (via LIN from steering wheel controls)
("Gateway_72", 10), # From J533 CAN gateway (aggregated data)
("Motor_14", 10), # From J623 Engine control module
("Airbag_02", 5), # From J234 Airbag control module
("Kombi_01", 2), # From J285 Instrument cluster
("Blinkmodi_02", 1), # From J519 BCM (sent at 1Hz when no lights active, 50Hz when active)
("Kombi_03", 0), # From J285 instrument cluster (not present on older cars, 1Hz when present)
]
if CP.transmissionType == TransmissionType.automatic:
messages.append(("Getriebe_11", 20)) # From J743 Auto transmission control module
elif CP.transmissionType == TransmissionType.direct:
messages.append(("EV_Gearshift", 10)) # From J??? unknown EV control module
if CP.networkLocation == NetworkLocation.fwdCamera:
# Radars are here on CANBUS.pt
messages += MqbExtraSignals.fwd_radar_messages
if CP.enableBsm:
messages += MqbExtraSignals.bsm_radar_messages
return CANParser(DBC[CP.carFingerprint]["pt"], messages, CANBUS.pt)
@staticmethod
def get_cam_can_parser(CP):
if CP.flags & VolkswagenFlags.PQ:
return CarState.get_cam_can_parser_pq(CP)
messages = []
if CP.flags & VolkswagenFlags.STOCK_HCA_PRESENT:
messages += [
("HCA_01", 1), # From R242 Driver assistance camera, 50Hz if steering/1Hz if not
]
if CP.networkLocation == NetworkLocation.fwdCamera:
messages += [
# sig_address, frequency
("LDW_02", 10) # From R242 Driver assistance camera
]
else:
# Radars are here on CANBUS.cam
messages += MqbExtraSignals.fwd_radar_messages
if CP.enableBsm:
messages += MqbExtraSignals.bsm_radar_messages
return CANParser(DBC[CP.carFingerprint]["pt"], messages, CANBUS.cam)
@staticmethod
def get_can_parser_pq(CP):
messages = [
# sig_address, frequency
("Bremse_1", 100), # From J104 ABS/ESP controller
("Bremse_3", 100), # From J104 ABS/ESP controller
("Lenkhilfe_3", 100), # From J500 Steering Assist with integrated sensors
("Lenkwinkel_1", 100), # From J500 Steering Assist with integrated sensors
("Motor_3", 100), # From J623 Engine control module
("Airbag_1", 50), # From J234 Airbag control module
("Bremse_5", 50), # From J104 ABS/ESP controller
("GRA_Neu", 50), # From J??? steering wheel control buttons
("Kombi_1", 50), # From J285 Instrument cluster
("Motor_2", 50), # From J623 Engine control module
("Motor_5", 50), # From J623 Engine control module
("Lenkhilfe_2", 20), # From J500 Steering Assist with integrated sensors
("Gate_Komf_1", 10), # From J533 CAN gateway
]
if CP.transmissionType == TransmissionType.automatic:
messages += [("Getriebe_1", 100)] # From J743 Auto transmission control module
elif CP.transmissionType == TransmissionType.manual:
messages += [("Motor_1", 100)] # From J623 Engine control module
if CP.networkLocation == NetworkLocation.fwdCamera:
# Extended CAN devices other than the camera are here on CANBUS.pt
messages += PqExtraSignals.fwd_radar_messages
if CP.enableBsm:
messages += PqExtraSignals.bsm_radar_messages
return CANParser(DBC[CP.carFingerprint]["pt"], messages, CANBUS.pt)
@staticmethod
def get_cam_can_parser_pq(CP):
messages = []
if CP.networkLocation == NetworkLocation.fwdCamera:
messages += [
# sig_address, frequency
("LDW_Status", 10) # From R242 Driver assistance camera
]
if CP.networkLocation == NetworkLocation.gateway:
# Radars are here on CANBUS.cam
messages += PqExtraSignals.fwd_radar_messages
if CP.enableBsm:
messages += PqExtraSignals.bsm_radar_messages
return CANParser(DBC[CP.carFingerprint]["pt"], messages, CANBUS.cam)
class MqbExtraSignals:
# Additional signal and message lists for optional or bus-portable controllers
fwd_radar_messages = [
("ACC_06", 50), # From J428 ACC radar control module
("ACC_10", 50), # From J428 ACC radar control module
("ACC_02", 17), # From J428 ACC radar control module
]
bsm_radar_messages = [
("SWA_01", 20), # From J1086 Lane Change Assist
]
class PqExtraSignals:
# Additional signal and message lists for optional or bus-portable controllers
fwd_radar_messages = [
("ACC_System", 50), # From J428 ACC radar control module
("ACC_GRA_Anzeige", 25), # From J428 ACC radar control module
]
bsm_radar_messages = [
("SWA_1", 20), # From J1086 Lane Change Assist
]

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,138 @@
from cereal import car
from panda import Panda
from openpilot.selfdrive.car import get_safety_config
from openpilot.selfdrive.car.interfaces import CarInterfaceBase
from openpilot.selfdrive.car.volkswagen.values import CAR, CANBUS, NetworkLocation, TransmissionType, GearShifter, VolkswagenFlags
ButtonType = car.CarState.ButtonEvent.Type
EventName = car.CarEvent.EventName
class CarInterface(CarInterfaceBase):
def __init__(self, CP, CarController, CarState):
super().__init__(CP, CarController, CarState)
if CP.networkLocation == NetworkLocation.fwdCamera:
self.ext_bus = CANBUS.pt
self.cp_ext = self.cp
else:
self.ext_bus = CANBUS.cam
self.cp_ext = self.cp_cam
self.eps_timer_soft_disable_alert = False
@staticmethod
def _get_params(ret, params, candidate: CAR, fingerprint, car_fw, disable_openpilot_long, experimental_long, docs):
ret.carName = "volkswagen"
ret.radarUnavailable = True
if ret.flags & VolkswagenFlags.PQ:
# Set global PQ35/PQ46/NMS parameters
ret.safetyConfigs = [get_safety_config(car.CarParams.SafetyModel.volkswagenPq)]
ret.enableBsm = 0x3BA in fingerprint[0] # SWA_1
if 0x440 in fingerprint[0] or docs: # Getriebe_1
ret.transmissionType = TransmissionType.automatic
else:
ret.transmissionType = TransmissionType.manual
if any(msg in fingerprint[1] for msg in (0x1A0, 0xC2)): # Bremse_1, Lenkwinkel_1
ret.networkLocation = NetworkLocation.gateway
else:
ret.networkLocation = NetworkLocation.fwdCamera
# The PQ port is in dashcam-only mode due to a fixed six-minute maximum timer on HCA steering. An unsupported
# EPS flash update to work around this timer, and enable steering down to zero, is available from:
# https://github.com/pd0wm/pq-flasher
# It is documented in a four-part blog series:
# https://blog.willemmelching.nl/carhacking/2022/01/02/vw-part1/
# Panda ALLOW_DEBUG firmware required.
ret.dashcamOnly = True
else:
# Set global MQB parameters
ret.safetyConfigs = [get_safety_config(car.CarParams.SafetyModel.volkswagen)]
ret.enableBsm = 0x30F in fingerprint[0] # SWA_01
if 0xAD in fingerprint[0] or docs: # Getriebe_11
ret.transmissionType = TransmissionType.automatic
elif 0x187 in fingerprint[0]: # EV_Gearshift
ret.transmissionType = TransmissionType.direct
else:
ret.transmissionType = TransmissionType.manual
if any(msg in fingerprint[1] for msg in (0x40, 0x86, 0xB2, 0xFD)): # Airbag_01, LWI_01, ESP_19, ESP_21
ret.networkLocation = NetworkLocation.gateway
else:
ret.networkLocation = NetworkLocation.fwdCamera
if 0x126 in fingerprint[2]: # HCA_01
ret.flags |= VolkswagenFlags.STOCK_HCA_PRESENT.value
# Global lateral tuning defaults, can be overridden per-vehicle
ret.steerLimitTimer = 0.4
if ret.flags & VolkswagenFlags.PQ:
ret.steerActuatorDelay = 0.2
CarInterfaceBase.configure_torque_tune(candidate, ret.lateralTuning)
else:
ret.steerActuatorDelay = 0.1
ret.lateralTuning.pid.kpBP = [0.]
ret.lateralTuning.pid.kiBP = [0.]
ret.lateralTuning.pid.kf = 0.00006
ret.lateralTuning.pid.kpV = [0.6]
ret.lateralTuning.pid.kiV = [0.2]
# Global longitudinal tuning defaults, can be overridden per-vehicle
ret.experimentalLongitudinalAvailable = ret.networkLocation == NetworkLocation.gateway or docs
if experimental_long:
# Proof-of-concept, prep for E2E only. No radar points available. Panda ALLOW_DEBUG firmware required.
ret.openpilotLongitudinalControl = True
ret.safetyConfigs[0].safetyParam |= Panda.FLAG_VOLKSWAGEN_LONG_CONTROL
if ret.transmissionType == TransmissionType.manual:
ret.minEnableSpeed = 4.5
ret.pcmCruise = not ret.openpilotLongitudinalControl
ret.stoppingControl = True
ret.stopAccel = -0.55
ret.vEgoStarting = 0.1
ret.vEgoStopping = 0.5
ret.longitudinalTuning.kpV = [0.1]
ret.longitudinalTuning.kiV = [0.0]
ret.autoResumeSng = ret.minEnableSpeed == -1
return ret
# returns a car.CarState
def _update(self, c, frogpilot_variables):
ret = self.CS.update(self.cp, self.cp_cam, self.cp_ext, self.CP.transmissionType, frogpilot_variables)
events = self.create_common_events(ret, extra_gears=[GearShifter.eco, GearShifter.sport, GearShifter.manumatic],
pcm_enable=not self.CS.CP.openpilotLongitudinalControl,
enable_buttons=(ButtonType.setCruise, ButtonType.resumeCruise))
# Low speed steer alert hysteresis logic
if self.CP.minSteerSpeed > 0. and ret.vEgo < (self.CP.minSteerSpeed + 1.):
self.low_speed_alert = True
elif ret.vEgo > (self.CP.minSteerSpeed + 2.):
self.low_speed_alert = False
if self.low_speed_alert:
events.add(EventName.belowSteerSpeed)
if self.CS.CP.openpilotLongitudinalControl:
if ret.vEgo < self.CP.minEnableSpeed + 0.5:
events.add(EventName.belowEngageSpeed)
if c.enabled and ret.vEgo < self.CP.minEnableSpeed:
events.add(EventName.speedTooLow)
if self.eps_timer_soft_disable_alert:
events.add(EventName.steerTimeLimit)
ret.events = events.to_msg()
return ret
def apply(self, c, now_nanos, frogpilot_variables):
new_actuators, can_sends, self.eps_timer_soft_disable_alert = self.CC.update(c, self.CS, self.ext_bus, now_nanos, frogpilot_variables)
return new_actuators, can_sends

View File

@@ -0,0 +1,137 @@
def create_steering_control(packer, bus, apply_steer, lkas_enabled):
values = {
"HCA_01_Status_HCA": 5 if lkas_enabled else 3,
"HCA_01_LM_Offset": abs(apply_steer),
"HCA_01_LM_OffSign": 1 if apply_steer < 0 else 0,
"HCA_01_Vib_Freq": 18,
"HCA_01_Sendestatus": 1 if lkas_enabled else 0,
"EA_ACC_Wunschgeschwindigkeit": 327.36,
}
return packer.make_can_msg("HCA_01", bus, values)
def create_eps_update(packer, bus, eps_stock_values, ea_simulated_torque):
values = {s: eps_stock_values[s] for s in [
"COUNTER", # Sync counter value to EPS output
"EPS_Lenkungstyp", # EPS rack type
"EPS_Berechneter_LW", # Absolute raw steering angle
"EPS_VZ_BLW", # Raw steering angle sign
"EPS_HCA_Status", # EPS HCA control status
]}
values.update({
# Absolute driver torque input and sign, with EA inactivity mitigation
"EPS_Lenkmoment": abs(ea_simulated_torque),
"EPS_VZ_Lenkmoment": 1 if ea_simulated_torque < 0 else 0,
})
return packer.make_can_msg("LH_EPS_03", bus, values)
def create_lka_hud_control(packer, bus, ldw_stock_values, enabled, steering_pressed, hud_alert, hud_control, lat_active):
values = {}
if len(ldw_stock_values):
values = {s: ldw_stock_values[s] for s in [
"LDW_SW_Warnung_links", # Blind spot in warning mode on left side due to lane departure
"LDW_SW_Warnung_rechts", # Blind spot in warning mode on right side due to lane departure
"LDW_Seite_DLCTLC", # Direction of most likely lane departure (left or right)
"LDW_DLC", # Lane departure, distance to line crossing
"LDW_TLC", # Lane departure, time to line crossing
]}
values.update({
"LDW_Status_LED_gelb": 1 if lat_active and steering_pressed else 0,
"LDW_Status_LED_gruen": 1 if lat_active and not steering_pressed else 0,
"LDW_Lernmodus_links": 3 if hud_control.leftLaneDepart else 1 + hud_control.leftLaneVisible,
"LDW_Lernmodus_rechts": 3 if hud_control.rightLaneDepart else 1 + hud_control.rightLaneVisible,
"LDW_Texte": hud_alert,
})
return packer.make_can_msg("LDW_02", bus, values)
def create_acc_buttons_control(packer, bus, gra_stock_values, cancel=False, resume=False):
values = {s: gra_stock_values[s] for s in [
"GRA_Hauptschalter", # ACC button, on/off
"GRA_Typ_Hauptschalter", # ACC main button type
"GRA_Codierung", # ACC button configuration/coding
"GRA_Tip_Stufe_2", # unknown related to stalk type
"GRA_ButtonTypeInfo", # unknown related to stalk type
]}
values.update({
"COUNTER": (gra_stock_values["COUNTER"] + 1) % 16,
"GRA_Abbrechen": cancel,
"GRA_Tip_Wiederaufnahme": resume,
})
return packer.make_can_msg("GRA_ACC_01", bus, values)
def acc_control_value(main_switch_on, acc_faulted, long_active):
if acc_faulted:
acc_control = 6
elif long_active:
acc_control = 3
elif main_switch_on:
acc_control = 2
else:
acc_control = 0
return acc_control
def acc_hud_status_value(main_switch_on, acc_faulted, long_active):
# TODO: happens to resemble the ACC control value for now, but extend this for init/gas override later
return acc_control_value(main_switch_on, acc_faulted, long_active)
def create_acc_accel_control(packer, bus, acc_type, acc_enabled, accel, acc_control, stopping, starting, esp_hold):
commands = []
acc_06_values = {
"ACC_Typ": acc_type,
"ACC_Status_ACC": acc_control,
"ACC_StartStopp_Info": acc_enabled,
"ACC_Sollbeschleunigung_02": accel if acc_enabled else 3.01,
"ACC_zul_Regelabw_unten": 0.2, # TODO: dynamic adjustment of comfort-band
"ACC_zul_Regelabw_oben": 0.2, # TODO: dynamic adjustment of comfort-band
"ACC_neg_Sollbeschl_Grad_02": 4.0 if acc_enabled else 0, # TODO: dynamic adjustment of jerk limits
"ACC_pos_Sollbeschl_Grad_02": 4.0 if acc_enabled else 0, # TODO: dynamic adjustment of jerk limits
"ACC_Anfahren": starting,
"ACC_Anhalten": stopping,
}
commands.append(packer.make_can_msg("ACC_06", bus, acc_06_values))
if starting:
acc_hold_type = 4 # hold release / startup
elif esp_hold:
acc_hold_type = 3 # hold standby
elif stopping:
acc_hold_type = 1 # hold request
else:
acc_hold_type = 0
acc_07_values = {
"ACC_Anhalteweg": 0.3 if stopping else 20.46, # Distance to stop (stopping coordinator handles terminal roll-out)
"ACC_Freilauf_Info": 2 if acc_enabled else 0,
"ACC_Folgebeschl": 3.02, # Not using secondary controller accel unless and until we understand its impact
"ACC_Sollbeschleunigung_02": accel if acc_enabled else 3.01,
"ACC_Anforderung_HMS": acc_hold_type,
"ACC_Anfahren": starting,
"ACC_Anhalten": stopping,
}
commands.append(packer.make_can_msg("ACC_07", bus, acc_07_values))
return commands
def create_acc_hud_control(packer, bus, acc_hud_status, set_speed, lead_distance, distance):
values = {
"ACC_Status_Anzeige": acc_hud_status,
"ACC_Wunschgeschw_02": set_speed if set_speed < 250 else 327.36,
"ACC_Gesetzte_Zeitluecke": distance + 2,
"ACC_Display_Prio": 3,
"ACC_Abstandsindex": lead_distance,
}
return packer.make_can_msg("ACC_02", bus, values)

105
selfdrive/car/volkswagen/pqcan.py Executable file
View File

@@ -0,0 +1,105 @@
def create_steering_control(packer, bus, apply_steer, lkas_enabled):
values = {
"LM_Offset": abs(apply_steer),
"LM_OffSign": 1 if apply_steer < 0 else 0,
"HCA_Status": 5 if (lkas_enabled and apply_steer != 0) else 3,
"Vib_Freq": 16,
}
return packer.make_can_msg("HCA_1", bus, values)
def create_lka_hud_control(packer, bus, ldw_stock_values, enabled, steering_pressed, hud_alert, hud_control, lat_active):
values = {}
if len(ldw_stock_values):
values = {s: ldw_stock_values[s] for s in [
"LDW_SW_Warnung_links", # Blind spot in warning mode on left side due to lane departure
"LDW_SW_Warnung_rechts", # Blind spot in warning mode on right side due to lane departure
"LDW_Seite_DLCTLC", # Direction of most likely lane departure (left or right)
"LDW_DLC", # Lane departure, distance to line crossing
"LDW_TLC", # Lane departure, time to line crossing
]}
values.update({
"LDW_Lampe_gelb": 1 if lat_active and steering_pressed else 0,
"LDW_Lampe_gruen": 1 if lat_active and not steering_pressed else 0,
"LDW_Lernmodus_links": 3 if hud_control.leftLaneDepart else 1 + hud_control.leftLaneVisible,
"LDW_Lernmodus_rechts": 3 if hud_control.rightLaneDepart else 1 + hud_control.rightLaneVisible,
"LDW_Textbits": hud_alert,
})
return packer.make_can_msg("LDW_Status", bus, values)
def create_acc_buttons_control(packer, bus, gra_stock_values, cancel=False, resume=False):
values = {s: gra_stock_values[s] for s in [
"GRA_Hauptschalt", # ACC button, on/off
"GRA_Typ_Hauptschalt", # ACC button, momentary vs latching
"GRA_Kodierinfo", # ACC button, configuration
"GRA_Sender", # ACC button, CAN message originator
]}
values.update({
"COUNTER": (gra_stock_values["COUNTER"] + 1) % 16,
"GRA_Abbrechen": cancel,
"GRA_Recall": resume,
})
return packer.make_can_msg("GRA_Neu", bus, values)
def acc_control_value(main_switch_on, acc_faulted, long_active):
if long_active:
acc_control = 1
elif main_switch_on:
acc_control = 2
else:
acc_control = 0
return acc_control
def acc_hud_status_value(main_switch_on, acc_faulted, long_active):
if acc_faulted:
hud_status = 6
elif long_active:
hud_status = 3
elif main_switch_on:
hud_status = 2
else:
hud_status = 0
return hud_status
def create_acc_accel_control(packer, bus, acc_type, acc_enabled, accel, acc_control, stopping, starting, esp_hold):
commands = []
values = {
"ACS_Sta_ADR": acc_control,
"ACS_StSt_Info": acc_enabled,
"ACS_Typ_ACC": acc_type,
"ACS_Anhaltewunsch": acc_type == 1 and stopping,
"ACS_FreigSollB": acc_enabled,
"ACS_Sollbeschl": accel if acc_enabled else 3.01,
"ACS_zul_Regelabw": 0.2 if acc_enabled else 1.27,
"ACS_max_AendGrad": 3.0 if acc_enabled else 5.08,
}
commands.append(packer.make_can_msg("ACC_System", bus, values))
return commands
def create_acc_hud_control(packer, bus, acc_hud_status, set_speed, lead_distance, distance):
values = {
"ACA_StaACC": acc_hud_status,
"ACA_Zeitluecke": distance + 2,
"ACA_V_Wunsch": set_speed,
"ACA_gemZeitl": lead_distance,
"ACA_PrioDisp": 3,
# TODO: restore dynamic pop-to-foreground/highlight behavior with ACA_PrioDisp and ACA_AnzDisplay
# TODO: ACA_kmh_mph handling probably needed to resolve rounding errors in displayed setpoint
}
return packer.make_can_msg("ACC_GRA_Anzeige", bus, values)

View File

@@ -0,0 +1,4 @@
from openpilot.selfdrive.car.interfaces import RadarInterfaceBase
class RadarInterface(RadarInterfaceBase):
pass

View File

@@ -0,0 +1,420 @@
from collections import namedtuple
from dataclasses import dataclass, field
from enum import Enum, IntFlag
from cereal import car
from panda.python import uds
from opendbc.can.can_define import CANDefine
from openpilot.common.conversions import Conversions as CV
from openpilot.selfdrive.car import dbc_dict, CarSpecs, DbcDict, PlatformConfig, Platforms
from openpilot.selfdrive.car.docs_definitions import CarFootnote, CarHarness, CarDocs, CarParts, Column, \
Device
from openpilot.selfdrive.car.fw_query_definitions import FwQueryConfig, Request, p16
Ecu = car.CarParams.Ecu
NetworkLocation = car.CarParams.NetworkLocation
TransmissionType = car.CarParams.TransmissionType
GearShifter = car.CarState.GearShifter
Button = namedtuple('Button', ['event_type', 'can_addr', 'can_msg', 'values'])
class CarControllerParams:
STEER_STEP = 2 # HCA_01/HCA_1 message frequency 50Hz
ACC_CONTROL_STEP = 2 # ACC_06/ACC_07/ACC_System frequency 50Hz
# Documented lateral limits: 3.00 Nm max, rate of change 5.00 Nm/sec.
# MQB vs PQ maximums are shared, but rate-of-change limited differently
# based on safety requirements driven by lateral accel testing.
STEER_MAX = 300 # Max heading control assist torque 3.00 Nm
STEER_DRIVER_MULTIPLIER = 3 # weight driver torque heavily
STEER_DRIVER_FACTOR = 1 # from dbc
STEER_TIME_MAX = 360 # Max time that EPS allows uninterrupted HCA steering control
STEER_TIME_ALERT = STEER_TIME_MAX - 10 # If mitigation fails, time to soft disengage before EPS timer expires
STEER_TIME_STUCK_TORQUE = 1.9 # EPS limits same torque to 6 seconds, reset timer 3x within that period
ACCEL_MAX = 2.0 # 2.0 m/s max acceleration
ACCEL_MAX_PLUS = 4.0 # 4.0 m/s max acceleration
ACCEL_MIN = -3.5 # 3.5 m/s max deceleration
def __init__(self, CP):
can_define = CANDefine(DBC[CP.carFingerprint]["pt"])
if CP.flags & VolkswagenFlags.PQ:
self.LDW_STEP = 5 # LDW_1 message frequency 20Hz
self.ACC_HUD_STEP = 4 # ACC_GRA_Anzeige frequency 25Hz
self.STEER_DRIVER_ALLOWANCE = 80 # Driver intervention threshold 0.8 Nm
self.STEER_DELTA_UP = 6 # Max HCA reached in 1.00s (STEER_MAX / (50Hz * 1.00))
self.STEER_DELTA_DOWN = 10 # Min HCA reached in 0.60s (STEER_MAX / (50Hz * 0.60))
if CP.transmissionType == TransmissionType.automatic:
self.shifter_values = can_define.dv["Getriebe_1"]["Waehlhebelposition__Getriebe_1_"]
self.hca_status_values = can_define.dv["Lenkhilfe_2"]["LH2_Sta_HCA"]
self.BUTTONS = [
Button(car.CarState.ButtonEvent.Type.setCruise, "GRA_Neu", "GRA_Neu_Setzen", [1]),
Button(car.CarState.ButtonEvent.Type.resumeCruise, "GRA_Neu", "GRA_Recall", [1]),
Button(car.CarState.ButtonEvent.Type.accelCruise, "GRA_Neu", "GRA_Up_kurz", [1]),
Button(car.CarState.ButtonEvent.Type.decelCruise, "GRA_Neu", "GRA_Down_kurz", [1]),
Button(car.CarState.ButtonEvent.Type.cancel, "GRA_Neu", "GRA_Abbrechen", [1]),
Button(car.CarState.ButtonEvent.Type.gapAdjustCruise, "GRA_Neu", "GRA_Zeitluecke", [1]),
]
self.LDW_MESSAGES = {
"none": 0, # Nothing to display
"laneAssistUnavail": 1, # "Lane Assist currently not available."
"laneAssistUnavailSysError": 2, # "Lane Assist system error"
"laneAssistUnavailNoSensorView": 3, # "Lane Assist not available. No sensor view."
"laneAssistTakeOver": 4, # "Lane Assist: Please Take Over Steering"
"laneAssistDeactivTrailer": 5, # "Lane Assist: no function with trailer"
}
else:
self.LDW_STEP = 10 # LDW_02 message frequency 10Hz
self.ACC_HUD_STEP = 6 # ACC_02 message frequency 16Hz
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))
self.STEER_DELTA_DOWN = 10 # Min HCA reached in 0.60s (STEER_MAX / (50Hz * 0.60))
if CP.transmissionType == TransmissionType.automatic:
self.shifter_values = can_define.dv["Getriebe_11"]["GE_Fahrstufe"]
elif CP.transmissionType == TransmissionType.direct:
self.shifter_values = can_define.dv["EV_Gearshift"]["GearPosition"]
self.hca_status_values = can_define.dv["LH_EPS_03"]["EPS_HCA_Status"]
self.BUTTONS = [
Button(car.CarState.ButtonEvent.Type.setCruise, "GRA_ACC_01", "GRA_Tip_Setzen", [1]),
Button(car.CarState.ButtonEvent.Type.resumeCruise, "GRA_ACC_01", "GRA_Tip_Wiederaufnahme", [1]),
Button(car.CarState.ButtonEvent.Type.accelCruise, "GRA_ACC_01", "GRA_Tip_Hoch", [1]),
Button(car.CarState.ButtonEvent.Type.decelCruise, "GRA_ACC_01", "GRA_Tip_Runter", [1]),
Button(car.CarState.ButtonEvent.Type.cancel, "GRA_ACC_01", "GRA_Abbrechen", [1]),
Button(car.CarState.ButtonEvent.Type.gapAdjustCruise, "GRA_ACC_01", "GRA_Verstellung_Zeitluecke", [1]),
]
self.LDW_MESSAGES = {
"none": 0, # Nothing to display
"laneAssistUnavailChime": 1, # "Lane Assist currently not available." with chime
"laneAssistUnavailNoSensorChime": 3, # "Lane Assist not available. No sensor view." with chime
"laneAssistTakeOverUrgent": 4, # "Lane Assist: Please Take Over Steering" with urgent beep
"emergencyAssistUrgent": 6, # "Emergency Assist: Please Take Over Steering" with urgent beep
"laneAssistTakeOverChime": 7, # "Lane Assist: Please Take Over Steering" with chime
"laneAssistTakeOver": 8, # "Lane Assist: Please Take Over Steering" silent
"emergencyAssistChangingLanes": 9, # "Emergency Assist: Changing lanes..." with urgent beep
"laneAssistDeactivated": 10, # "Lane Assist deactivated." silent with persistent icon afterward
}
class CANBUS:
pt = 0
cam = 2
class VolkswagenFlags(IntFlag):
# Detected flags
STOCK_HCA_PRESENT = 1
# Static flags
PQ = 2
@dataclass
class VolkswagenMQBPlatformConfig(PlatformConfig):
dbc_dict: DbcDict = field(default_factory=lambda: dbc_dict('vw_mqb_2010', None))
@dataclass
class VolkswagenPQPlatformConfig(PlatformConfig):
dbc_dict: DbcDict = field(default_factory=lambda: dbc_dict('vw_golf_mk4', None))
def init(self):
self.flags |= VolkswagenFlags.PQ
@dataclass(frozen=True, kw_only=True)
class VolkswagenCarSpecs(CarSpecs):
centerToFrontRatio: float = 0.45
steerRatio: float = 15.6
class Footnote(Enum):
KAMIQ = CarFootnote(
"Not including the China market Kamiq, which is based on the (currently) unsupported PQ34 platform.",
Column.MODEL)
PASSAT = CarFootnote(
"Refers only to the MQB-based European B8 Passat, not the NMS Passat in the USA/China/Mideast markets.",
Column.MODEL)
SKODA_HEATED_WINDSHIELD = CarFootnote(
"Some Škoda vehicles are equipped with heated windshields, which are known " +
"to block GPS signal needed for some comma 3X functionality.",
Column.MODEL)
VW_EXP_LONG = CarFootnote(
"Only available for vehicles using a gateway (J533) harness. At this time, vehicles using a camera harness " +
"are limited to using stock ACC.",
Column.LONGITUDINAL)
VW_MQB_A0 = CarFootnote(
"Model-years 2022 and beyond may have a combined CAN gateway and BCM, which is supported by openpilot " +
"in software, but doesn't yet have a harness available from the comma store.",
Column.HARDWARE)
@dataclass
class VWCarDocs(CarDocs):
package: str = "Adaptive Cruise Control (ACC) & Lane Assist"
car_parts: CarParts = field(default_factory=CarParts.common([CarHarness.j533]))
def init_make(self, CP: car.CarParams):
self.footnotes.append(Footnote.VW_EXP_LONG)
if "SKODA" in CP.carFingerprint:
self.footnotes.append(Footnote.SKODA_HEATED_WINDSHIELD)
if CP.carFingerprint in (CAR.CRAFTER_MK2, CAR.TRANSPORTER_T61):
self.car_parts = CarParts([Device.threex_angled_mount, CarHarness.j533])
# Check the 7th and 8th characters of the VIN before adding a new CAR. If the
# chassis code is already listed below, don't add a new CAR, just add to the
# FW_VERSIONS for that existing CAR.
# Exception: SEAT Leon and SEAT Ateca share a chassis code
class CAR(Platforms):
ARTEON_MK1 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN ARTEON 1ST GEN", # Chassis AN
[
VWCarDocs("Volkswagen Arteon 2018-23", video_link="https://youtu.be/FAomFKPFlDA"),
VWCarDocs("Volkswagen Arteon R 2020-23", video_link="https://youtu.be/FAomFKPFlDA"),
VWCarDocs("Volkswagen Arteon eHybrid 2020-23", video_link="https://youtu.be/FAomFKPFlDA"),
VWCarDocs("Volkswagen CC 2018-22", video_link="https://youtu.be/FAomFKPFlDA"),
],
VolkswagenCarSpecs(mass=1733, wheelbase=2.84),
)
ATLAS_MK1 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN ATLAS 1ST GEN", # Chassis CA
[
VWCarDocs("Volkswagen Atlas 2018-23"),
VWCarDocs("Volkswagen Atlas Cross Sport 2020-22"),
VWCarDocs("Volkswagen Teramont 2018-22"),
VWCarDocs("Volkswagen Teramont Cross Sport 2021-22"),
VWCarDocs("Volkswagen Teramont X 2021-22"),
],
VolkswagenCarSpecs(mass=2011, wheelbase=2.98),
)
CADDY_MK3 = VolkswagenPQPlatformConfig(
"VOLKSWAGEN CADDY 3RD GEN", # Chassis 2K
[
VWCarDocs("Volkswagen Caddy 2019"),
VWCarDocs("Volkswagen Caddy Maxi 2019"),
],
VolkswagenCarSpecs(mass=1613, wheelbase=2.6, minSteerSpeed=21 * CV.KPH_TO_MS),
)
CRAFTER_MK2 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN CRAFTER 2ND GEN", # Chassis SY/SZ
[
VWCarDocs("Volkswagen Crafter 2017-23", video_link="https://youtu.be/4100gLeabmo"),
VWCarDocs("Volkswagen e-Crafter 2018-23", video_link="https://youtu.be/4100gLeabmo"),
VWCarDocs("Volkswagen Grand California 2019-23", video_link="https://youtu.be/4100gLeabmo"),
VWCarDocs("MAN TGE 2017-23", video_link="https://youtu.be/4100gLeabmo"),
VWCarDocs("MAN eTGE 2020-23", video_link="https://youtu.be/4100gLeabmo"),
],
VolkswagenCarSpecs(mass=2100, wheelbase=3.64, minSteerSpeed=50 * CV.KPH_TO_MS),
)
GOLF_MK7 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN GOLF 7TH GEN", # Chassis 5G/AU/BA/BE
[
VWCarDocs("Volkswagen e-Golf 2014-20"),
VWCarDocs("Volkswagen Golf 2015-20", auto_resume=False),
VWCarDocs("Volkswagen Golf Alltrack 2015-19", auto_resume=False),
VWCarDocs("Volkswagen Golf GTD 2015-20"),
VWCarDocs("Volkswagen Golf GTE 2015-20"),
VWCarDocs("Volkswagen Golf GTI 2015-21", auto_resume=False),
VWCarDocs("Volkswagen Golf R 2015-19"),
VWCarDocs("Volkswagen Golf SportsVan 2015-20"),
],
VolkswagenCarSpecs(mass=1397, wheelbase=2.62),
)
JETTA_MK7 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN JETTA 7TH GEN", # Chassis BU
[
VWCarDocs("Volkswagen Jetta 2018-24"),
VWCarDocs("Volkswagen Jetta GLI 2021-24"),
],
VolkswagenCarSpecs(mass=1328, wheelbase=2.71),
)
PASSAT_MK8 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN PASSAT 8TH GEN", # Chassis 3G
[
VWCarDocs("Volkswagen Passat 2015-22", footnotes=[Footnote.PASSAT]),
VWCarDocs("Volkswagen Passat Alltrack 2015-22"),
VWCarDocs("Volkswagen Passat GTE 2015-22"),
],
VolkswagenCarSpecs(mass=1551, wheelbase=2.79),
)
PASSAT_NMS = VolkswagenPQPlatformConfig(
"VOLKSWAGEN PASSAT NMS", # Chassis A3
[VWCarDocs("Volkswagen Passat NMS 2017-22")],
VolkswagenCarSpecs(mass=1503, wheelbase=2.80, minSteerSpeed=50*CV.KPH_TO_MS, minEnableSpeed=20*CV.KPH_TO_MS),
)
POLO_MK6 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN POLO 6TH GEN", # Chassis AW
[
VWCarDocs("Volkswagen Polo 2018-23", footnotes=[Footnote.VW_MQB_A0]),
VWCarDocs("Volkswagen Polo GTI 2018-23", footnotes=[Footnote.VW_MQB_A0]),
],
VolkswagenCarSpecs(mass=1230, wheelbase=2.55),
)
SHARAN_MK2 = VolkswagenPQPlatformConfig(
"VOLKSWAGEN SHARAN 2ND GEN", # Chassis 7N
[
VWCarDocs("Volkswagen Sharan 2018-22"),
VWCarDocs("SEAT Alhambra 2018-20"),
],
VolkswagenCarSpecs(mass=1639, wheelbase=2.92, minSteerSpeed=50*CV.KPH_TO_MS),
)
TAOS_MK1 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN TAOS 1ST GEN", # Chassis B2
[VWCarDocs("Volkswagen Taos 2022-23")],
VolkswagenCarSpecs(mass=1498, wheelbase=2.69),
)
TCROSS_MK1 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN T-CROSS 1ST GEN", # Chassis C1
[VWCarDocs("Volkswagen T-Cross 2021", footnotes=[Footnote.VW_MQB_A0])],
VolkswagenCarSpecs(mass=1150, wheelbase=2.60),
)
TIGUAN_MK2 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN TIGUAN 2ND GEN", # Chassis AD/BW
[
VWCarDocs("Volkswagen Tiguan 2018-24"),
VWCarDocs("Volkswagen Tiguan eHybrid 2021-23"),
],
VolkswagenCarSpecs(mass=1715, wheelbase=2.74),
)
TOURAN_MK2 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN TOURAN 2ND GEN", # Chassis 1T
[VWCarDocs("Volkswagen Touran 2016-23")],
VolkswagenCarSpecs(mass=1516, wheelbase=2.79),
)
TRANSPORTER_T61 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN TRANSPORTER T6.1", # Chassis 7H/7L
[
VWCarDocs("Volkswagen Caravelle 2020"),
VWCarDocs("Volkswagen California 2021-23"),
],
VolkswagenCarSpecs(mass=1926, wheelbase=3.00, minSteerSpeed=14.0),
)
TROC_MK1 = VolkswagenMQBPlatformConfig(
"VOLKSWAGEN T-ROC 1ST GEN", # Chassis A1
[VWCarDocs("Volkswagen T-Roc 2018-22", footnotes=[Footnote.VW_MQB_A0])],
VolkswagenCarSpecs(mass=1413, wheelbase=2.63),
)
AUDI_A3_MK3 = VolkswagenMQBPlatformConfig(
"AUDI A3 3RD GEN", # Chassis 8V/FF
[
VWCarDocs("Audi A3 2014-19"),
VWCarDocs("Audi A3 Sportback e-tron 2017-18"),
VWCarDocs("Audi RS3 2018"),
VWCarDocs("Audi S3 2015-17"),
],
VolkswagenCarSpecs(mass=1335, wheelbase=2.61),
)
AUDI_Q2_MK1 = VolkswagenMQBPlatformConfig(
"AUDI Q2 1ST GEN", # Chassis GA
[VWCarDocs("Audi Q2 2018")],
VolkswagenCarSpecs(mass=1205, wheelbase=2.61),
)
AUDI_Q3_MK2 = VolkswagenMQBPlatformConfig(
"AUDI Q3 2ND GEN", # Chassis 8U/F3/FS
[VWCarDocs("Audi Q3 2019-23")],
VolkswagenCarSpecs(mass=1623, wheelbase=2.68),
)
SEAT_ATECA_MK1 = VolkswagenMQBPlatformConfig(
"SEAT ATECA 1ST GEN", # Chassis 5F
[VWCarDocs("SEAT Ateca 2018")],
VolkswagenCarSpecs(mass=1900, wheelbase=2.64),
)
SEAT_LEON_MK3 = VolkswagenMQBPlatformConfig(
"SEAT LEON 3RD GEN", # Chassis 5F
[VWCarDocs("SEAT Leon 2014-20")],
VolkswagenCarSpecs(mass=1227, wheelbase=2.64),
)
SKODA_FABIA_MK4 = VolkswagenMQBPlatformConfig(
"SKODA FABIA 4TH GEN", # Chassis PJ
[VWCarDocs("Škoda Fabia 2022-23", footnotes=[Footnote.VW_MQB_A0])],
VolkswagenCarSpecs(mass=1266, wheelbase=2.56),
)
SKODA_KAMIQ_MK1 = VolkswagenMQBPlatformConfig(
"SKODA KAMIQ 1ST GEN", # Chassis NW
[VWCarDocs("Škoda Kamiq 2021-23", footnotes=[Footnote.VW_MQB_A0, Footnote.KAMIQ])],
VolkswagenCarSpecs(mass=1265, wheelbase=2.66),
)
SKODA_KAROQ_MK1 = VolkswagenMQBPlatformConfig(
"SKODA KAROQ 1ST GEN", # Chassis NU
[VWCarDocs("Škoda Karoq 2019-23")],
VolkswagenCarSpecs(mass=1278, wheelbase=2.66),
)
SKODA_KODIAQ_MK1 = VolkswagenMQBPlatformConfig(
"SKODA KODIAQ 1ST GEN", # Chassis NS
[VWCarDocs("Škoda Kodiaq 2017-23")],
VolkswagenCarSpecs(mass=1569, wheelbase=2.79),
)
SKODA_OCTAVIA_MK3 = VolkswagenMQBPlatformConfig(
"SKODA OCTAVIA 3RD GEN", # Chassis NE
[
VWCarDocs("Škoda Octavia 2015-19"),
VWCarDocs("Škoda Octavia RS 2016"),
],
VolkswagenCarSpecs(mass=1388, wheelbase=2.68),
)
SKODA_SCALA_MK1 = VolkswagenMQBPlatformConfig(
"SKODA SCALA 1ST GEN", # Chassis NW
[VWCarDocs("Škoda Scala 2020-23", footnotes=[Footnote.VW_MQB_A0])],
VolkswagenCarSpecs(mass=1192, wheelbase=2.65),
)
SKODA_SUPERB_MK3 = VolkswagenMQBPlatformConfig(
"SKODA SUPERB 3RD GEN", # Chassis 3V/NP
[VWCarDocs("Škoda Superb 2015-22")],
VolkswagenCarSpecs(mass=1505, wheelbase=2.84),
)
# All supported cars should return FW from the engine, srs, eps, and fwdRadar. Cars
# with a manual trans won't return transmission firmware, but all other cars will.
#
# The 0xF187 SW part number query should return in the form of N[NX][NX] NNN NNN [X[X]],
# where N=number, X=letter, and the trailing two letters are optional. Performance
# tuners sometimes tamper with that field (e.g. 8V0 9C0 BB0 1 from COBB/EQT). Tampered
# ECU SW part numbers are invalid for vehicle ID and compatibility checks. Try to have
# them repaired by the tuner before including them in openpilot.
VOLKSWAGEN_VERSION_REQUEST_MULTI = bytes([uds.SERVICE_TYPE.READ_DATA_BY_IDENTIFIER]) + \
p16(uds.DATA_IDENTIFIER_TYPE.VEHICLE_MANUFACTURER_SPARE_PART_NUMBER) + \
p16(uds.DATA_IDENTIFIER_TYPE.VEHICLE_MANUFACTURER_ECU_SOFTWARE_VERSION_NUMBER) + \
p16(uds.DATA_IDENTIFIER_TYPE.APPLICATION_DATA_IDENTIFICATION)
VOLKSWAGEN_VERSION_RESPONSE = bytes([uds.SERVICE_TYPE.READ_DATA_BY_IDENTIFIER + 0x40])
VOLKSWAGEN_RX_OFFSET = 0x6a
FW_QUERY_CONFIG = FwQueryConfig(
# TODO: add back whitelists after we gather enough data
requests=[request for bus, obd_multiplexing in [(1, True), (1, False), (0, False)] for request in [
Request(
[VOLKSWAGEN_VERSION_REQUEST_MULTI],
[VOLKSWAGEN_VERSION_RESPONSE],
# whitelist_ecus=[Ecu.srs, Ecu.eps, Ecu.fwdRadar],
rx_offset=VOLKSWAGEN_RX_OFFSET,
bus=bus,
logging=(bus != 1 or not obd_multiplexing),
obd_multiplexing=obd_multiplexing,
),
Request(
[VOLKSWAGEN_VERSION_REQUEST_MULTI],
[VOLKSWAGEN_VERSION_RESPONSE],
# whitelist_ecus=[Ecu.engine, Ecu.transmission],
bus=bus,
logging=(bus != 1 or not obd_multiplexing),
obd_multiplexing=obd_multiplexing,
),
]],
extra_ecus=[(Ecu.fwdCamera, 0x74f, None)],
)
DBC = CAR.create_dbc_map()