IQ.Pilot Release Commit @ 3807439

This commit is contained in:
IQ.Lvbs CI [bot]
2026-07-21 19:07:37 -05:00
parent ccb06b3624
commit 0952a162ef
98 changed files with 6193 additions and 1808 deletions

View File

@@ -0,0 +1,74 @@
<!--- AUTOGENERATED FROM selfdrive/car/CARS_template.md, DO NOT EDIT. --->
# Support Information for {{all_car_docs | length}} Known Cars
|{{ExtraCarsColumn | map(attribute='value') | join('|') | replace(hardware_col_name, wide_hardware_col_name)}}|
|---|---|---|{% for _ in range((ExtraCarsColumn | length) - 3) %}{{':---:|'}}{% endfor +%}
{% for car_docs in all_car_docs %}
|{% for column in ExtraCarsColumn %}{{car_docs.get_extra_cars_column(column)}}|{% endfor %}
{% endfor %}
# Types of Support
**iqdbc can support many more cars than it currently does.** There are a few reasons your car may not be supported.
If your car doesn't fit into any of the incompatibility criteria here, then there's a good chance it can be supported!
We're adding support for new cars all the time. **We don't have a roadmap for car support**, and in fact, most car
support comes from users like you!
## Upstream
A supported vehicle is one that just works when you install a comma device. All supported cars provide a better
experience than any stock system. Supported vehicles reference the US market unless otherwise specified.
## Under Review
A vehicle under review is one for which software support has been merged into upstream openpilot, but hasn't yet been
tested for drive quality and conformance with [comma safety guidelines](https://github.com/commaai/openpilot/blob/master/docs/SAFETY.md).
This is a normal part of the development and quality assurance process. This vehicle will not work when upstream
openpilot is installed, but custom forks may allow their use.
## Custom
Vehicles in this category are not considered plug-and-play. Software support is included in upstream openpilot, but
these vehicles might not have a harness in the comma store, or the physical install might be at an unusual or cumbersome
location, or they might need unusual configuration after install. These vehicles will not work with release builds of
openpilot, but depending on the situation, development builds or custom forks may allow their use.
### SecOC cars with recoverable keys
For a small subset of SecOC-protected vehicles, tools may be available in the community to recover the SecOC keys. These
tools, and the recovery process, are not part of openpilot. If supplied with a valid SecOC key, development builds or
custom forks may work with these vehicles. Release builds of openpilot don't support SecOC.
## Dashcam
Dashcam vehicles have software support in upstream openpilot, but will go into "dashcam mode" at startup and will not
engage. This may be due to known issues with driving safety or quality, or it may be a work in progress that isn't yet
ready for safety and quality review.
## Community
Although they're not upstream, the community has openpilot running on other makes and models. See the 'Community
Supported Models' section of each make [on our wiki](https://wiki.comma.ai/).
Some notable works-in-progress:
* Honda
* 2022-24 Acura RDX, commaai/iqdbc#1967
* Camera ACC stability improvements, commaai/iqdbc#2192
* Alpha longitudinal stability improvements, commaai/iqdbc#2347 and commaai/iqdbc#2165
## Incompatible
### CAN Bus Security
Vehicles with CAN security measures, such as AUTOSAR Secure Onboard Communication (SecOC) are not usable with openpilot
unless the owner can recover the message signing key and implement CAN message signing. Examples include certain newer
Toyota, and the GM Global B platform.
### FlexRay
All the cars that openpilot supports use a [CAN bus](https://en.wikipedia.org/wiki/CAN_bus) for communication between all the car's computers, however a
CAN bus isn't the only way that the computers in your car can communicate. Most, if not all, vehicles from the following
manufacturers use [FlexRay](https://en.wikipedia.org/wiki/FlexRay) instead of a CAN bus: **BMW, Mercedes, Audi, Land Rover, and some Volvo**. These cars
may one day be supported, but we have no immediate plans to support FlexRay.

View File

@@ -535,9 +535,14 @@ struct CarParams {
steerLimitAlert @28 :Bool;
steerLimitTimer @47 :Float32; # time before steerLimitAlert is issued
vEgoStopping @29 :Float32; # Speed at which the car goes into stopping state
vEgoStarting @59 :Float32; # Speed at which the car goes into starting state
steerControlType @34 :SteerControlType;
radarUnavailable @35 :Bool; # True when radar objects aren't visible on CAN or aren't parsed out
stopAccel @60 :Float32; # Required acceleration to keep vehicle stationary
stoppingDecelRate @52 :Float32; # m/s^2/s while trying to stop
startAccel @32 :Float32; # Required acceleration to get car moving
startingState @70 :Bool; # Does this car make use of special starting state
steerActuatorDelay @36 :Float32; # Steering wheel actuator delay in seconds
longitudinalActuatorDelay @58 :Float32; # Gas/Brake actuator delay in seconds
@@ -770,9 +775,4 @@ struct CarParams {
stoppingControlDEPRECATED @31 :Bool; # Does the car allow full control even at lows speeds when stopping
radarTimeStepDEPRECATED @45: Float32 = 0.05; # time delta between radar updates, 20Hz is very standard
enableDsuDEPRECATED @5 :Bool; # driving support unit
vEgoStartingDEPRECATED @59 :Float32;
startAccelDEPRECATED @32 :Float32;
startingStateDEPRECATED @70 :Bool;
vEgoStoppingDEPRECATED @29 :Float32;
stoppingDecelRateDEPRECATED @52 :Float32;
}

View File

@@ -87,7 +87,7 @@ def fingerprint(can_recv: CanRecvCallable, can_send: CanSendCallable, set_obd_mu
cached_params: CarParamsT | None,
fixed_fingerprint: str | None) -> tuple[str | None, dict, str, list[CarParams.CarFw], CarParams.FingerprintSource, bool]:
fixed_fingerprint = fixed_fingerprint or os.environ.get('FINGERPRINT', "")
skip_fw_query = os.environ.get('SKIP_FW_QUERY', False) or bool(fixed_fingerprint)
skip_fw_query = os.environ.get('SKIP_FW_QUERY', False)
disable_fw_cache = os.environ.get('DISABLE_FW_CACHE', False)
ecu_rx_addrs = set()

View File

@@ -21,6 +21,7 @@ from iqdbc.car.extra_cars import CAR as EXTRA
EXTRA_CARS_MD_OUT = os.path.join(BASEDIR, "../", "../", "docs", "CARS.md")
EXTRA_CARS_MD_TEMPLATE = os.path.join(BASEDIR, "CARS_template.md")
# TODO: merge these platforms into normal car ports with SupportType flag
ExtraPlatform = Platform | EXTRA
@@ -105,6 +106,7 @@ if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Auto generates supportability info docs for all known cars",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("--template", default=EXTRA_CARS_MD_TEMPLATE, help="Override default template filename")
parser.add_argument("--out", default=EXTRA_CARS_MD_OUT, help="Override default generated filename")
args = parser.parse_args()

View File

@@ -125,6 +125,9 @@ class CarInterface(CarInterfaceBase, CarInterfaceExt):
# Tuning for experimental long
ret.longitudinalTuning.kiV = [2.0, 1.5]
ret.stoppingDecelRate = 2.0 # reach brake quickly after enabling
ret.vEgoStopping = 0.25
ret.vEgoStarting = 0.25
if alpha_long:
ret.pcmCruise = False

View File

@@ -23,10 +23,6 @@ MAX_ANGLE_FRAMES = 89
MAX_ANGLE_CONSECUTIVE_FRAMES = 2
def _use_stock_lkas_request_path(CP) -> bool:
return CP.carFingerprint == CAR.HYUNDAI_PALISADE
def process_hud_alert(enabled, fingerprint, hud_control):
sys_warning = (hud_control.visualAlert in (VisualAlert.steerRequired, VisualAlert.ldw))
@@ -96,12 +92,12 @@ class CarController(CarControllerBase, EsccCarController, LeadDataCarController,
else:
self.apply_torque_last = apply_torque
if not _use_stock_lkas_request_path(self.CP) and apply_torque == 0 and CC.latActive and CS.out.steeringPressed:
if apply_torque == 0 and CC.latActive and CS.out.steeringPressed:
apply_steer_req = False
# Hold torque with induced temporary fault when cutting the actuation bit
# FIXME: we don't use this with CAN FD?
torque_fault = False if _use_stock_lkas_request_path(self.CP) else (CC.latActive and not apply_steer_req)
torque_fault = CC.latActive and not apply_steer_req
# accel + longitudinal
accel = float(np.clip(actuators.accel, CarControllerParams.ACCEL_MIN, CarControllerParams.ACCEL_MAX))

View File

@@ -7,10 +7,6 @@ from iqdbc.iqpilot.car.hyundai.lead_data_ext import CanLeadData
hyundai_checksum = crcmod.mkCrcFun(0x11D, initCrc=0xFD, rev=False, xorOut=0xdf)
def _use_stock_scc_surrogates(CP) -> bool:
return CP.carFingerprint == CAR.HYUNDAI_PALISADE
def create_lkas11(packer, frame, CP, apply_torque, steer_req,
torque_fault, lkas11, sys_warning, sys_state, enabled,
left_lane, right_lane,
@@ -141,7 +137,7 @@ def create_acc_commands(packer, enabled, accel, upper_jerk, idx, lead_data: CanL
commands = []
def get_scc11_values():
values = {
return {
"MainMode_ACC": 1 if main_cruise_enabled else 0,
"TauGapSet": hud_control.leadDistanceBars,
"VSetDis": set_speed if enabled else 0,
@@ -152,11 +148,6 @@ def create_acc_commands(packer, enabled, accel, upper_jerk, idx, lead_data: CanL
"ACC_ObjRelSpd": lead_data.lead_rel_speed,
"ACC_ObjDist": int(lead_data.lead_distance), # close lead makes controls tighter
}
if _use_stock_scc_surrogates(CP):
values["ObjValid"] = 1
values["ACC_ObjStatus"] = 1
values["ACC_ObjDist"] = 1
return values
def get_scc12_values():
scc12_values = {
@@ -185,17 +176,15 @@ def create_acc_commands(packer, enabled, accel, upper_jerk, idx, lead_data: CanL
return values
def get_scc14_values():
values = {
return {
"ComfortBandUpper": tuning.comfort_band_upper, # stock usually is 0 but sometimes uses higher values
"ComfortBandLower": tuning.comfort_band_lower, # stock usually is 0 but sometimes uses higher values
"JerkUpperLimit": tuning.jerk_upper, # stock usually is 1.0 but sometimes uses higher values
"JerkLowerLimit": tuning.jerk_lower, # stock usually is 0.5 but sometimes uses higher values
"ACCMode": 2 if enabled and long_override else 1 if enabled else 4, # stock will always be 4 instead of 0 after first disengage
"ObjGap": lead_data.object_gap, # 5: >30, m, 4: 25-30 m, 3: 20-25 m, 2: < 20 m, 0: no lead
"ObjDistStat": lead_data.object_rel_gap,
}
if not _use_stock_scc_surrogates(CP):
values["ObjDistStat"] = lead_data.object_rel_gap
return values
def get_fca11_values():
return {

View File

@@ -134,6 +134,9 @@ class CarInterface(CarInterfaceBase):
ret.radarUnavailable = RADAR_START_ADDR not in fingerprint[1] or Bus.radar not in DBC[ret.carFingerprint]
ret.openpilotLongitudinalControl = alpha_long and ret.alphaLongitudinalAvailable
ret.pcmCruise = not ret.openpilotLongitudinalControl
ret.startingState = True
ret.vEgoStarting = 0.1
ret.startAccel = 1.0
ret.longitudinalActuatorDelay = 0.5
if ret.openpilotLongitudinalControl:

View File

@@ -1,61 +0,0 @@
from iqdbc.can import CANParser, CANPacker
from iqdbc.car import Bus
from iqdbc.car.hyundai import hyundaican
from iqdbc.car.hyundai.values import CAR, DBC
class DummyHudControl:
leadDistanceBars = 3
leadVisible = True
class DummyLeadData:
lead_visible = False
lead_rel_speed = -7
lead_distance = 42
object_gap = 5
object_rel_gap = 2
class DummyTuning:
comfort_band_upper = 0.2
comfort_band_lower = 0.3
jerk_upper = 1.7
jerk_lower = 1.2
desired_accel = 0.4
actual_accel = 0.3
stopping = False
class DummyCP:
carFingerprint = CAR.HYUNDAI_PALISADE
flags = 0
def _decode(msg_name: str, addr: int, dat: bytes):
cp = CANParser(DBC[CAR.HYUNDAI_PALISADE][Bus.pt], [(msg_name, 0)], 0)
cp.update([(0, [(addr, dat, 0)])])
return cp.vl[msg_name]
def test_palisade_uses_stock_scc_surrogates():
packer = CANPacker(DBC[CAR.HYUNDAI_PALISADE][Bus.pt])
cp = DummyCP()
hud = DummyHudControl()
tuning = DummyTuning()
lead = DummyLeadData()
scc11 = hyundaican.create_acc_commands(
packer, True, 0.5, 3.0, 1, lead, hud, 72, False, False, True, cp, False, tuning
)[0]
scc14 = hyundaican.create_acc_commands(
packer, True, 0.5, 3.0, 1, lead, hud, 72, False, False, True, cp, False, tuning
)[2]
scc11_vals = _decode("SCC11", scc11[0], scc11[1])
assert scc11_vals["ObjValid"] == 1
assert scc11_vals["ACC_ObjStatus"] == 1
assert scc11_vals["ACC_ObjDist"] == 1
scc14_vals = _decode("SCC14", scc14[0], scc14[1])
assert "ObjDistStat" not in scc14_vals or scc14_vals["ObjDistStat"] == 0

View File

@@ -253,6 +253,9 @@ class CarInterfaceBase(ABC, CarInterfaceBaseIQ):
ret.steerRatioRear = 0. # no rear steering, at least on the listed cars aboveA
ret.openpilotLongitudinalControl = False
ret.stopAccel = -2.0
ret.stoppingDecelRate = 0.8 # brake_travel/s while trying to stop
ret.vEgoStopping = 0.5
ret.vEgoStarting = 0.5
ret.longitudinalTuning.kpBP = [0.]
ret.longitudinalTuning.kpV = [0.]
ret.longitudinalTuning.kiBP = [0.]

View File

@@ -6,6 +6,7 @@ from iqdbc.car.vehicle_model import VehicleModel
FRICTION_THRESHOLD = 0.2
# ISO 11270
ISO_LATERAL_ACCEL = 3.0 # m/s^2
ISO_LATERAL_JERK = 5.0 # m/s^3
AVERAGE_ROAD_ROLL = 0.06 # ~3.4 degrees

View File

@@ -3,7 +3,7 @@ import os
import capnp
import urllib.parse
import warnings
from urllib.request import urlopen, Request
from urllib.request import urlopen
import zstandard as zstd
from iqdbc.car.common.basedir import BASEDIR
@@ -27,7 +27,7 @@ class LogReader:
_, ext = os.path.splitext(urllib.parse.urlparse(fn).path)
if fn.startswith("http"):
with urlopen(Request(fn, headers={"User-Agent": "iqdbc"})) as f:
with urlopen(fn) as f:
dat = f.read()
else:
with open(fn, "rb") as f:

View File

@@ -32,6 +32,7 @@ class CarInterface(CarInterfaceBase):
ret.safetyConfigs[0].safetyParam |= RivianSafetyFlags.LONG_CONTROL.value
ret.longitudinalActuatorDelay = 0.35
ret.vEgoStopping = 0.25
ret.stopAccel = 0
return ret

View File

@@ -1,2 +0,0 @@
# FIXME: gate by FingerPrint
TESLA_BLINKERS = False

View File

@@ -3,13 +3,11 @@ from iqdbc.can import CANPacker
from iqdbc.car import Bus
from iqdbc.car.lateral import apply_steer_angle_limits_vm
from iqdbc.car.interfaces import CarControllerBase
from iqdbc.car.tesla import TESLA_BLINKERS
from iqdbc.car.tesla.teslacan import TeslaCAN
from iqdbc.car.tesla.values import CarControllerParams
from iqdbc.car.vehicle_model import VehicleModel
from openpilot.iqpilot.selfdrive.car.enhanced_stock_longitudinal_control import get_set_speed_kph_from_params
from iqdbc.iqpilot.car.tesla.coop_steering import CoopSteeringCarController
from iqdbc.iqpilot.car.tesla.values import TeslaFlagsIQ
def get_safety_CP():
@@ -30,11 +28,6 @@ class CarController(CarControllerBase):
# Vehicle model used for lateral limiting
self.VM = VehicleModel(get_safety_CP())
self.has_vehicle_bus = bool(CP_IQ.flags & TeslaFlagsIQ.HAS_VEHICLE_BUS)
self.body_controls_counter_last = -1
self.blinker_request_prev = False
self.blinker_cancel_frame = 0
def update(self, CC, CC_IQ, CS, now_nanos):
actuators = CC.actuators
can_sends = []
@@ -59,8 +52,6 @@ class CarController(CarControllerBase):
if self.frame % 4 == 0:
state = 13 if CC.cruiseControl.cancel or CS.das_accCancel else 4 # 4=ACC_ON, 13=ACC_CANCEL_GENERIC_SILENT
accel = float(np.clip(actuators.accel, CarControllerParams.ACCEL_MIN, CarControllerParams.ACCEL_MAX))
if not CC.longActive:
accel = 0.
cntr = (self.frame // 4) % 8
set_speed_kph = get_set_speed_kph_from_params(CC_IQ.params)
can_sends.append(self.tesla_can.create_longitudinal_command(state, accel, cntr, CS.out.vEgo, CC.longActive,
@@ -72,30 +63,6 @@ class CarController(CarControllerBase):
cntr = (CS.das_control["DAS_controlCounter"] + 1) % 8
can_sends.append(self.tesla_can.create_longitudinal_command(13, 0, cntr, CS.out.vEgo, False, True))
# Nav blinker control via DAS_bodyControls on the vehicle bus, phase-locked to the car's
# counter. Cancel on the trailing edge since the body controller latches the signal.
stock_dat = getattr(CS, 'das_body_controls_dat', b"")
# FIXME: gate by FingerPrint
if TESLA_BLINKERS and self.has_vehicle_bus and len(stock_dat) >= 8:
left_blinker = CC.leftBlinker
right_blinker = CC.rightBlinker
driver_opposes = (left_blinker and CS.out.rightBlinker) or (right_blinker and CS.out.leftBlinker)
if driver_opposes:
left_blinker = right_blinker = False
nav_requesting = left_blinker or right_blinker
if self.blinker_request_prev and not nav_requesting and not driver_opposes:
self.blinker_cancel_frame = self.frame + 150 # ~1.5 s
self.blinker_request_prev = nav_requesting
cancel = not nav_requesting and not driver_opposes and self.frame < self.blinker_cancel_frame
body_counter = stock_dat[6] >> 4
if body_counter != self.body_controls_counter_last:
can_sends.append(self.tesla_can.create_body_controls(stock_dat, left_blinker, right_blinker, cancel))
self.body_controls_counter_last = body_counter
# TODO: HUD control
new_actuators = actuators.as_builder()
new_actuators.steeringAngleDeg = self.apply_angle_last

View File

@@ -1,17 +1,13 @@
import copy
from iqdbc.can import CANDefine, CANParser
from iqdbc.car import Bus, create_button_events, structs
from iqdbc.car.carlog import carlog
from iqdbc.car.common.conversions import Conversions as CV
from iqdbc.car.interfaces import CarStateBase
from iqdbc.car.tesla import TESLA_BLINKERS
from iqdbc.car.tesla.teslacan import get_steer_ctrl_type
from iqdbc.car.tesla.values import DBC, CANBUS, GEAR_MAP, STEER_THRESHOLD, TeslaFlags
from iqdbc.iqpilot.car.tesla.carstate_ext import CarStateExt
from iqdbc.iqpilot.car.tesla.values import TeslaFlagsIQ
from openpilot.common.params import Params
from openpilot.system.proprietary_runtime._verified_import import import_verified_module
vehicle_state = import_verified_module("iqpilot_alc_private", "iqpilot_private.konn3kt.iqlvbs.vehicle_state")
ButtonType = structs.CarState.ButtonEvent.Type
@@ -26,12 +22,13 @@ class CarState(CarStateBase, CarStateExt):
self.summon = False
self.summon_prev = False
self.cruise_enabled_prev = False
self.fsd14_error_logged = False
self.suspected_fsd14 = False
self.suspected_fsd14_clear_frames = 0
self.hands_on_level = 0
self.acc_state_last = 0
self.das_control = None
self.das_body_controls_dat = b""
self._odometer_store = vehicle_state.VehicleOdometerStore(CP, Params())
self.cruise_override = False
def update_summon_state(self, summon_state: str, cruise_enabled: bool):
@@ -139,40 +136,54 @@ class CarState(CarStateBase, CarStateExt):
ret.stockAeb = cp_ap_party.vl["DAS_control"]["DAS_aebEvent"] == 1
# LKAS
steer_control_type = int(cp_ap_party.vl["DAS_steeringControl"]["DAS_steeringControlType"])
if self.CP.flags & TeslaFlags.LEGACY_DAS_STEERING:
steer_control_type >>= 1 # legacy firmware uses a 2-bit field, one bit up from the 3-bit signal
ret.stockLkas = steer_control_type == 2 # LANE_KEEP_ASSIST
# On FSD 14+, ANGLE_CONTROL behavior changed to allow user winddown while actuating.
# FSD switched from using ANGLE_CONTROL to LANE_KEEP_ASSIST to likely keep the old steering override disengage logic.
# LKAS switched from LANE_KEEP_ASSIST to ANGLE_CONTROL to likely allow overriding LKAS events smoothly
lkas_ctrl_type = get_steer_ctrl_type(self.CP.flags, 2)
ret.stockLkas = cp_ap_party.vl["DAS_steeringControl"]["DAS_steeringControlType"] == lkas_ctrl_type # LANE_KEEP_ASSIST
# Stock Autosteer should be disengaged (includes FSD)
# TODO: find for TESLA_MODEL_X and HW2.5 vehicles
if not (self.CP.flags & TeslaFlags.MISSING_DAS_SETTINGS):
ret.invalidLkasSetting = cp_ap_party.vl["DAS_status"]["DAS_autopilotState"] not in (0, 1, 2) # DISABLED, UNAVAILABLE, AVAILABLE
# Because we don't have FSD 14 detection outside of a set of FW, we should check if this FW is accidentally missing from FSD_14_FW
# 1. If in Autosteer or FSD, already caught by invalidLkasSetting
# 2. If in TACC and DAS ever sends ANGLE_CONTROL (1), we can infer it's trying to do LKAS on FSD 14+
# NOTE: Tesla's latest firmware changed ELDA (Emergency Lane Departure Assist) to use ANGLE_CONTROL (1)
# instead of EMERGENCY_LANE_KEEP (3). Exclude ELDA by checking eac_status so it doesn't latch suspected_fsd14.
eac_is_emergency = eac_status == "EMERGENCY_LANE_KEEP"
angle_control = cp_ap_party.vl["DAS_steeringControl"]["DAS_steeringControlType"] == 1 and not eac_is_emergency # ANGLE_CONTROL, excluding ELDA
if not ret.invalidLkasSetting and angle_control and not self.CP.flags & TeslaFlags.FSD_14:
self.suspected_fsd14 = True
self.suspected_fsd14_clear_frames = 0
if self.suspected_fsd14:
ret.invalidLkasSetting = True
if not self.fsd14_error_logged:
carlog.error("FSD 14 detected, but FW not in FSD_14_FW set")
self.fsd14_error_logged = True
# Un-latch if ANGLE_CONTROL has been absent for ~3 s (100 frames @ ~33 Hz).
# This allows re-engagement after transient triggers (e.g. if ELDA slips through on new FW variants).
if not angle_control:
self.suspected_fsd14_clear_frames += 1
if self.suspected_fsd14_clear_frames >= 100:
self.suspected_fsd14 = False
self.suspected_fsd14_clear_frames = 0
else:
self.suspected_fsd14_clear_frames = 0
# Buttons # ToDo: add Gap adjust button
# Messages needed by carcontroller
self.das_control = copy.copy(cp_ap_party.vl["DAS_control"])
# Raw stock DAS_bodyControls bytes (bus 2), used to ride the blinker on the vehicle bus.
# FIXME: gate by FingerPrint
if TESLA_BLINKERS and Bus.cam in can_parsers:
self.das_body_controls_dat = bytes(can_parsers[Bus.cam].dat.get(0x3E9, b""))
CarStateExt.update(self, ret, ret_iq, can_parsers)
if ret.odometer > 0.0:
ret.odometer = self._odometer_store.record(ret.odometer) or 0.0
return ret, ret_iq
@staticmethod
def get_can_parsers(CP, CP_IQ):
parsers = {
return {
Bus.party: CANParser(DBC[CP.carFingerprint][Bus.party], [], CANBUS.party),
Bus.ap_party: CANParser(DBC[CP.carFingerprint][Bus.party], [], CANBUS.autopilot_party),
**CarStateExt.get_parser(CP, CP_IQ),
}
# Stock DAS_bodyControls from the AP bus (bus 2) for the nav blinker.
if TESLA_BLINKERS and CP_IQ.flags & TeslaFlagsIQ.HAS_VEHICLE_BUS and Bus.adas in DBC[CP.carFingerprint]:
parsers[Bus.cam] = CANParser(DBC[CP.carFingerprint][Bus.adas], [("DAS_bodyControls", 2)], CANBUS.autopilot_party)
return parsers

View File

@@ -2,7 +2,7 @@ from iqdbc.car import Bus, get_safety_config, structs
from iqdbc.car.interfaces import CarInterfaceBase
from iqdbc.car.tesla.carcontroller import CarController
from iqdbc.car.tesla.carstate import CarState
from iqdbc.car.tesla.values import TeslaSafetyFlags, TeslaFlags, CANBUS, CAR, DBC, LEGACY_DAS_STEERING_FW, Ecu
from iqdbc.car.tesla.values import TeslaSafetyFlags, TeslaFlags, CANBUS, CAR, DBC, FSD_14_FW, Ecu
from iqdbc.car.tesla.radar_interface import RadarInterface, RADAR_START_ADDR
from iqdbc.iqpilot.car.tesla.values import TeslaFlagsIQ, TeslaSafetyFlagsIQ
@@ -41,10 +41,14 @@ class CarInterface(CarInterfaceBase):
ret.openpilotLongitudinalControl = True
ret.safetyConfigs[0].safetyParam |= TeslaSafetyFlags.LONG_CONTROL.value
legacy_das = any(fw.ecu == Ecu.eps and fw.fwVersion in LEGACY_DAS_STEERING_FW.get(candidate, []) for fw in car_fw)
if legacy_das:
ret.flags |= TeslaFlags.LEGACY_DAS_STEERING.value
ret.safetyConfigs[0].safetyParam |= TeslaSafetyFlags.LEGACY_DAS_STEERING.value
ret.vEgoStopping = 0.1
ret.vEgoStarting = 0.1
ret.stoppingDecelRate = 0.3
fsd_14 = any(fw.ecu == Ecu.eps and fw.fwVersion in FSD_14_FW.get(candidate, []) for fw in car_fw)
if fsd_14:
ret.flags |= TeslaFlags.FSD_14.value
ret.safetyConfigs[0].safetyParam |= TeslaSafetyFlags.FSD_14.value
ret.dashcamOnly = candidate in (CAR.TESLA_MODEL_X,) # dashcam only, pending find invalidLkasSetting signal
@@ -59,10 +63,7 @@ class CarInterface(CarInterfaceBase):
if candidate == CAR.TESLA_MODEL_X:
stock_cp.dashcamOnly = False
# Vehicle-bus messages can be slow enough to miss the initial capture window.
# Accept either the established 0x3DF marker or the absolute odometer frame.
vehicle_bus_seen = any(0x3DF in bus or 0x3B6 in bus for bus in fingerprint.values())
if vehicle_bus_seen:
if 0x3DF in fingerprint[1]:
ret.flags |= TeslaFlagsIQ.HAS_VEHICLE_BUS.value
ret.iqSafetyFlags |= TeslaSafetyFlagsIQ.HAS_VEHICLE_BUS

View File

@@ -3,6 +3,14 @@ from iqdbc.car.tesla.values import CANBUS, CarControllerParams, TeslaFlags
from iqdbc.car import DT_CTRL
def get_steer_ctrl_type(flags: int, ctrl_type: int) -> int:
# Returns the flipped signal value for DAS_steeringControlType on FSD 14
if flags & TeslaFlags.FSD_14:
return {1: 2, 2: 1}.get(ctrl_type, ctrl_type)
else:
return ctrl_type
class TeslaCAN:
def __init__(self, CP, packer):
self.CP = CP
@@ -10,15 +18,14 @@ class TeslaCAN:
self.l_jerk = 0.0
def create_steering_control(self, angle, enabled, control_type):
# control_type comes from coop_steering: ANGLE_CONTROL (1) normally, LANE_KEEP_ASSIST (2) when cooperative steering is enabled
control_type = control_type if enabled else 0
if self.CP.flags & TeslaFlags.LEGACY_DAS_STEERING:
control_type <<= 1 # legacy firmware uses a 2-bit field, one bit up from the 3-bit signal
# On FSD 14+, ANGLE_CONTROL behavior changed to allow user winddown while actuating.
# with openpilot, after overriding w/ ANGLE_CONTROL the wheel snaps back to the original angle abruptly
# so we now use LANE_KEEP_ASSIST to match stock FSD.
# see carstate.py for more details
values = {
"DAS_steeringAngleRequest": -angle,
"DAS_steeringHapticRequest": 0,
"DAS_steeringControlType": control_type,
"DAS_steeringControlType": get_steer_ctrl_type(self.CP.flags, control_type if enabled else 0),
}
return self.packer.make_can_msg("DAS_steeringControl", CANBUS.party, values)
@@ -53,32 +60,6 @@ class TeslaCAN:
return self.packer.make_can_msg("APS_eacMonitor", CANBUS.party, values)
def create_body_controls(self, stock_dat, left_blinker, right_blinker, cancel=False):
# Ride alongside the car's native DAS_bodyControls: copy the raw frame, override only the
# turn-indicator bits, and stamp counter + 1 so our frame supersedes the stock one.
dat = bytearray(stock_dat)
if len(dat) < 8:
dat.extend(b"\x00" * (8 - len(dat)))
if left_blinker or right_blinker:
turn_req = 1 if left_blinker else 2 # DAS_TURN_INDICATOR_LEFT / _RIGHT
dat[1] = (dat[1] & ~0x07) | (turn_req & 0x07)
dat[2] = (dat[2] & ~0x3C) | (1 << 2) # DAS_ACTIVE_NAV_LANE_CHANGE
elif cancel:
dat[1] = (dat[1] & ~0x07) | 0x03 # DAS_TURN_INDICATOR_CANCEL
dat[2] = (dat[2] & ~0x3C) | (4 << 2) # DAS_CANCEL_LANE_CHANGE
counter = (((dat[6] >> 4) + 1) & 0x0F)
dat[6] = (dat[6] & ~0xF0) | (counter << 4)
addr = 0x3E9
checksum = (addr & 0xFF) + ((addr >> 8) & 0xFF)
for i in range(7):
checksum += dat[i]
dat[7] = checksum & 0xFF
return addr, bytes(dat), CANBUS.vehicle
def tesla_checksum(address: int, sig, d: bytearray) -> int:
checksum = (address & 0xFF) + ((address >> 8) & 0xFF)

View File

@@ -4,7 +4,6 @@ from iqdbc.car.tesla.teslacan import TeslaCAN
from iqdbc.car.tesla.radar_interface import RADAR_START_ADDR
from iqdbc.car.tesla.carcontroller import CarController
from iqdbc.car.tesla.values import CAR
from iqdbc.can import CANPacker, CANParser
class TestTeslaFingerprint:
@@ -32,19 +31,6 @@ class TestTeslaCan:
def make_can_msg(self, name, bus, values):
return name, bus, values
def test_vehicle_bus_odometer_decodes_kilometers(self):
packer = CANPacker("tesla_model3_vehicle")
parser = CANParser("tesla_model3_vehicle", [("ID3B6UI_odometer", 1)], 1)
message = packer.make_can_msg("ID3B6UI_odometer", 1, {
"UI_odometer": 29150.377,
"UI_odometerCounter": 1,
"UI_odometerChecksum": 0,
})
parser.update([1_000_000_000, [message]])
assert parser.vl["ID3B6UI_odometer"]["UI_odometer"] == 29150.377
def test_longitudinal_command_does_not_reference_missing_jerk_attr(self):
CP = CarInterface.get_non_essential_params(CAR.TESLA_MODEL_3)
tesla_can = TeslaCAN(CP, self.DummyPacker())

View File

@@ -79,41 +79,16 @@ FW_QUERY_CONFIG = FwQueryConfig(
]
)
# Cars with this EPS FW have a 2-bit DAS_steeringControlType and use TeslaFlags.LEGACY_DAS_STEERING
LEGACY_DAS_STEERING_FW = {
# Cars with this EPS FW have FSD 14 and use TeslaFlags.FSD_14
FSD_14_FW = {
CAR.TESLA_MODEL_3: [
b'TeM3_E014p10_0.0.0 (16),E014.17.00',
b'TeM3_E014p10_0.0.0 (16),EL014.17.00',
b'TeM3_ES014p11_0.0.0 (25),ES014.19.0',
b'TeMYG4_DCS_Update_0.0.0 (13),E4014.28.1',
b'TeMYG4_DCS_Update_0.0.0 (9),E4014.26.0',
b'TeMYG4_Legacy3Y_0.0.0 (2),E4015.02.0',
b'TeMYG4_Legacy3Y_0.0.0 (5),E4015.03.2',
b'TeMYG4_Legacy3Y_0.0.0 (5),E4L015.03.2',
b'TeMYG4_Main_0.0.0 (59),E4H014.29.0',
b'TeMYG4_Main_0.0.0 (65),E4H015.01.0',
b'TeMYG4_Main_0.0.0 (67),E4H015.02.1',
b'TeMYG4_SingleECU_0.0.0 (33),E4S014.27',
b'TeMYG4_Main_0.0.0 (77),E4HP015.04.5',
b'TeMYG4_Main_0.0.0 (78),E4HP015.05.0',
],
CAR.TESLA_MODEL_Y: [
b'TeM3_E014p10_0.0.0 (16),Y002.18.00',
b'TeM3_E014p10_0.0.0 (16),YP002.18.00',
b'TeM3_ES014p11_0.0.0 (16),YS002.17',
b'TeM3_ES014p11_0.0.0 (25),YS002.19.0',
b'TeMYG4_DCS_Update_0.0.0 (13),Y4002.27.1',
b'TeMYG4_DCS_Update_0.0.0 (13),Y4P002.27.1',
b'TeMYG4_DCS_Update_0.0.0 (9),Y4P002.25.0',
b'TeMYG4_Legacy3Y_0.0.0 (2),Y4003.02.0',
b'TeMYG4_Legacy3Y_0.0.0 (2),Y4P003.02.0',
b'TeMYG4_Legacy3Y_0.0.0 (5),Y4003.03.2',
b'TeMYG4_Legacy3Y_0.0.0 (5),Y4P003.03.2',
b'TeMYG4_SingleECU_0.0.0 (28),Y4S002.23.0',
b'TeMYG4_SingleECU_0.0.0 (33),Y4S002.26',
],
CAR.TESLA_MODEL_X: [
b'TeM3_SP_XP002p2_0.0.0 (23),XPR003.6.0',
b'TeM3_SP_XP002p2_0.0.0 (36),XPR003.10.0',
],
b'TeMYG4_Legacy3Y_0.0.0 (6),Y4003.04.0',
b'TeMYG4_Main_0.0.0 (77),Y4003.05.4',
]
}
@@ -164,12 +139,12 @@ class CarControllerParams:
class TeslaSafetyFlags(IntFlag):
LONG_CONTROL = 1
LEGACY_DAS_STEERING = 2
FSD_14 = 2
class TeslaFlags(IntFlag):
LONG_CONTROL = 1
LEGACY_DAS_STEERING = 2
FSD_14 = 2
MISSING_DAS_SETTINGS = 4

View File

@@ -301,8 +301,7 @@ routes = [
CarTestRoute("578742b26807f756|00000010--41ee3e5bec", VOLKSWAGEN.VOLKSWAGEN_JETTA_MK6),
CarTestRoute("58a7d3b707987d65/2021-03-25--17-26-37", VOLKSWAGEN.VOLKSWAGEN_JETTA_MK7),
CarTestRoute("4d134e099430fba2/2021-03-26--00-26-06", VOLKSWAGEN.VOLKSWAGEN_PASSAT_MK8),
CarTestRoute("b29ee8c5a0a735d1|000000dc--a384e9083e", VOLKSWAGEN.VOLKSWAGEN_PASSAT_NMS, segment=0),
CarTestRoute("0f53129ed44f6920|00000287--3efbddeb96", VOLKSWAGEN.VOLKSWAGEN_PASSAT_NMS, segment=0),
CarTestRoute("3cfdec54aa035f3f/2022-07-19--23-45-10", VOLKSWAGEN.VOLKSWAGEN_PASSAT_NMS),
CarTestRoute("0cd0b7f7e31a3853/2021-11-03--19-30-22", VOLKSWAGEN.VOLKSWAGEN_POLO_MK6),
CarTestRoute("064d1816e448f8eb/2022-09-29--15-32-34", VOLKSWAGEN.VOLKSWAGEN_SHARAN_MK2),
CarTestRoute("7d82b2f3a9115f1f/2021-10-21--15-39-42", VOLKSWAGEN.VOLKSWAGEN_TAOS_MK1),

View File

@@ -1,133 +0,0 @@
#!/usr/bin/env python3
"""Real-CAN replay invariant tests for VW torque platforms (PQ / MQB / MLB).
Replays real konn3kt routes through the car interface with openpilot lateral
INACTIVE (latActive=False) and asserts openpilot never transmits an active-steering
HCA command (active status or non-zero torque). Re-transmitting the stock camera's
active HCA while not in control (stock-LKAS forwarding) leaves the EPS faulted for
the whole drive (LH2_Sta_HCA=FAULT) - the regression that bricked steering on a PQ
Passat NMS with a factory LKAS camera. Panda accepts these frames, so only a replay
invariant like this catches it.
Self-contained within iqdbc: routes are resolved through konn3kt's public
/v1/route/<id>/files endpoint (URLs are signed server-side, no auth/token needed).
"""
import json
import os
import urllib.parse
import urllib.request
from collections import Counter
import pytest
from iqdbc.can.parser import CANParser
from iqdbc.car import Bus, structs
from iqdbc.car.can_definitions import CanData
from iqdbc.car.car_helpers import can_fingerprint, interfaces
from iqdbc.car.logreader import LogReader
from iqdbc.car.volkswagen.values import CAR, DBC, VolkswagenFlags
API_HOST = os.environ.get("API_HOST", "https://api-iqlabs.konn3kt.com")
# konn3kt-hosted VW routes. (route_id, segment, platform, label)
# Add MQB routes here as konn3kt-hosted MQB logs become available.
VW_ROUTES = [
("b29ee8c5a0a735d1|000000dc--a384e9083e", 0, CAR.VOLKSWAGEN_PASSAT_NMS, "PQ with stock LKAS camera"),
("0f53129ed44f6920|00000287--3efbddeb96", 0, CAR.VOLKSWAGEN_PASSAT_NMS, "PQ without stock LKAS camera"),
]
# Per-platform HCA message: (address, msg, status signal, torque signal, active-status values)
HCA_INFO = {
"pq": (0xD2, "HCA_1", "HCA_Status", "LM_Offset", (5, 7)),
"mqb": (0x126, "HCA_01", "HCA_01_Status_HCA", "HCA_01_LM_Offset", (5, 6, 7)),
}
def _request_headers() -> dict[str, str]:
# konn3kt's edge rejects the default urllib User-Agent with 403. The routes are public
# (access returns early for public routes), but IQ.Pilot/Cabana tooling conventionally
# sends a Konn3kt user JWT, so include one when available (env or ~/.comma/auth.json).
headers = {"User-Agent": "iqdbc"}
token = os.environ.get("KONN3KT_ACCESS_TOKEN")
if not token:
try:
with open(os.path.expanduser("~/.comma/auth.json")) as f:
token = json.load(f).get("access_token")
except (OSError, ValueError):
token = None
if token:
headers["Authorization"] = f"JWT {token}"
return headers
def _rlog_url(route_id: str, segment: int) -> str:
req = urllib.request.Request(f"{API_HOST}/v1/route/{urllib.parse.quote(route_id, safe='|')}/files",
headers=_request_headers())
with urllib.request.urlopen(req, timeout=30) as f:
files = json.load(f)
for url in files.get("logs", []):
# path looks like /connectdata/<dongle>/<log>/<seg>/rlog.zst
parts = urllib.parse.urlparse(url).path.rstrip("/").split("/")
if len(parts) >= 2 and parts[-2] == str(segment):
return url
raise RuntimeError(f"no rlog for {route_id} segment {segment} (uploaded & public?)")
def _load_can(route_id: str, segment: int):
lr = LogReader(_rlog_url(route_id, segment), only_union_types=True, sort_by_time=True)
return [(m.logMonoTime, [CanData(c.address, c.dat, c.src) for c in m.can]) for m in lr if m.which() == "can"]
@pytest.mark.parametrize("route_id,segment,platform,label", VW_ROUTES)
def test_vw_inactive_steering_invariant(route_id, segment, platform, label):
can_msgs = _load_can(route_id, segment)
assert len(can_msgs) > 1000, f"insufficient CAN data for {label}: {len(can_msgs)} frames"
# fingerprint from a fresh iterator over the (unmutated) frame list
frame_iter = (frames for _, frames in can_msgs)
def can_recv(wait_for_one: bool = False):
return [next(frame_iter, [])]
_, fingerprint = can_fingerprint(can_recv)
CarInterface = interfaces[platform]
CP = CarInterface.get_params(platform, fingerprint, [], False, False, False)
CP_IQ = CarInterface.get_params_iq(CP, platform, fingerprint, [], False, False, False)
if CP.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO):
pytest.skip("invariant covers torque-based VW platforms (PQ/MQB/MLB)")
key = "pq" if CP.flags & VolkswagenFlags.PQ else "mqb"
hca_addr, hca_msg, status_sig, torque_sig, active_status = HCA_INFO[key]
cp = CANParser(DBC[platform][Bus.pt], [(hca_msg, 0)], 0)
CI = CarInterface(CP, CP_IQ)
CC = structs.CarControl().as_reader() # latActive defaults to False
CC_IQ = structs.IQCarControl()
hca_seen = 0
violations = Counter()
for i, (mono, frames) in enumerate(can_msgs):
CI.update([(mono, frames)])
_, sendcan = CI.apply(CC, CC_IQ, mono)
if i < 300: # CarController / CANParser warmup
continue
for addr, dat, bus in sendcan:
if addr != hca_addr or bus != 0:
continue
hca_seen += 1
cp.update([(mono, [(addr, bytes(dat), 0)])])
if int(cp.vl[hca_msg][status_sig]) in active_status:
violations["active_status"] += 1
if abs(cp.vl[hca_msg][torque_sig]) > 0:
violations["nonzero_torque"] += 1
assert hca_seen > 50, f"{label}: no HCA steering messages transmitted to inspect"
assert not len(violations), \
f"{label}: openpilot TX'd active HCA while latActive=False: {dict(violations)}"
if __name__ == "__main__":
import sys
sys.exit(pytest.main([__file__, "-v"]))

View File

@@ -67,7 +67,6 @@ legend = ["LAT_ACCEL_FACTOR", "MAX_LAT_ACCEL_MEASURED", "FRICTION"]
"VOLKSWAGEN_CADDY_MK3" = [1.2, 1.2, 0.1]
"VOLKSWAGEN_PASSAT_NMS" = [2.5, 2.5, 0.1]
"VOLKSWAGEN_SHARAN_MK2" = [2.5, 2.5, 0.1]
"SEAT_ALHAMBRA_MK1" = [2.5, 2.5, 0.1]
"HYUNDAI_SANTA_CRUZ_1ST_GEN" = [2.7, 2.7, 0.1]
"KIA_SPORTAGE_5TH_GEN" = [2.6, 2.6, 0.1]
"GENESIS_GV70_1ST_GEN" = [2.42, 2.42, 0.1]

View File

@@ -76,7 +76,6 @@ legend = ["LAT_ACCEL_FACTOR", "MAX_LAT_ACCEL_MEASURED", "FRICTION"]
"VOLKSWAGEN_JETTA_MK6" = "VOLKSWAGEN_PASSAT_NMS"
"VOLKSWAGEN_PASSAT_MK7" = "VOLKSWAGEN_PASSAT_NMS"
"VOLKSWAGEN_PASSAT_NMS_PLUS" = "VOLKSWAGEN_PASSAT_NMS"
"VOLKSWAGEN_PASSAT_B7" = "VOLKSWAGEN_PASSAT_NMS"
"SUBARU_CROSSTREK_HYBRID" = "SUBARU_IMPREZA_2020"
"SUBARU_FORESTER_HYBRID" = "SUBARU_IMPREZA_2020"

View File

@@ -260,7 +260,6 @@ class CarController(CarControllerBase, GasInterceptorCarController):
freeze_integrator=actuators.longControlState != LongCtrlState.pid)
else:
self.long_pid.reset()
pcm_accel_cmd = 0.
# Along with rate limiting positive jerk above, this greatly improves gas response time
# Consider the net acceleration request that the PCM should be applying (pitch included)
@@ -270,9 +269,7 @@ class CarController(CarControllerBase, GasInterceptorCarController):
elif net_acceleration_request_min > 0.3:
self.permit_braking = False
sdsu_tssp_long_active = bool(self.CP_IQ.flags & ToyotaFlagsIQ.SMART_DSU) and \
(self.CP.carFingerprint not in TSS2_CAR) and CC.longActive
pcm_accel_cmd = actuators.accel if sdsu_tssp_long_active else pcm_accel_cmd
pcm_accel_cmd = pcm_accel_cmd if self.CP.carFingerprint in TSS2_CAR else actuators.accel
pcm_accel_cmd = float(np.clip(pcm_accel_cmd, self.params.ACCEL_MIN, self.params.ACCEL_MAX))
main_accel_cmd = 0. if self.CP.flags & ToyotaFlags.SECOC.value else pcm_accel_cmd

View File

@@ -36,6 +36,7 @@ class CarInterface(CarInterfaceBase):
if ret.flags & ToyotaFlags.SECOC.value:
ret.secOcRequired = True
ret.safetyConfigs[0].safetyParam |= ToyotaSafetyFlags.SECOC.value
ret.dashcamOnly = is_release
if candidate in ANGLE_CONTROL_CAR:
ret.steerControlType = SteerControlType.angle
@@ -118,6 +119,10 @@ class CarInterface(CarInterfaceBase):
if candidate in TSS2_CAR:
ret.flags |= ToyotaFlags.RAISED_ACCEL_LIMIT.value
ret.vEgoStopping = 0.25
ret.vEgoStarting = 0.25
ret.stoppingDecelRate = 0.3 # reach stopping target smoothly
# Hybrids have much quicker longitudinal actuator response
if ret.flags & ToyotaFlags.HYBRID.value:
ret.longitudinalActuatorDelay = 0.05
@@ -127,6 +132,9 @@ class CarInterface(CarInterfaceBase):
@staticmethod
def _get_params_iq(stock_cp: structs.CarParams, ret: structs.IQCarParams, candidate, fingerprint: dict[int, dict[int, int]],
car_fw: list[structs.CarParams.CarFw], alpha_long: bool, is_release_iq: bool, docs: bool) -> structs.IQCarParams:
if stock_cp.flags & ToyotaFlags.SECOC.value and stock_cp.fingerprintSource == structs.CarParams.FingerprintSource.fixed:
stock_cp.dashcamOnly = False
if candidate in UNSUPPORTED_DSU_CAR:
ret.iqSafetyFlags |= ToyotaSafetyFlagsIQ.UNSUPPORTED_DSU

View File

@@ -4,20 +4,26 @@ from iqdbc.car.toyota.values import CAR, ToyotaFlags
from iqdbc.iqpilot.car.toyota.values import ToyotaFlagsIQ
def test_secoc_toyota_not_dashcam_on_release():
# SecOC Toyotas are controllable regardless of branch or fingerprint source.
def test_forced_secoc_toyota_clears_dashcam_mode():
candidate = CAR.TOYOTA_SIENNA_4TH_GEN
# is_release=True (release branch) must not force dashcam mode
cp = CarInterface.get_params(candidate, gen_empty_fingerprint(), [], False, True, False)
assert cp.flags & ToyotaFlags.SECOC.value
assert cp.secOcRequired
assert cp.dashcamOnly
cp.fingerprintSource = structs.CarParams.FingerprintSource.fixed
CarInterface.get_params_iq(cp, candidate, gen_empty_fingerprint(), [], False, True, False)
assert not cp.dashcamOnly
# fw/can-sourced fingerprint (not forced) must also stay controllable
def test_automatic_secoc_toyota_release_stays_dashcam():
candidate = CAR.TOYOTA_SIENNA_4TH_GEN
cp = CarInterface.get_params(candidate, gen_empty_fingerprint(), [], False, True, False)
assert cp.dashcamOnly
cp.fingerprintSource = structs.CarParams.FingerprintSource.can
CarInterface.get_params_iq(cp, candidate, gen_empty_fingerprint(), [], False, True, False)
assert not cp.dashcamOnly
assert cp.dashcamOnly
def test_smart_dsu_clears_disable_radar_on_radar_acc_toyota():

View File

@@ -14,16 +14,10 @@ from iqdbc.car.common.conversions import Conversions as CV
from iqdbc.car.common.numpy_fast import clip, interp
from iqdbc.car.interfaces import CarControllerBase
from iqdbc.car.volkswagen import mlbcan, mqbcan, pqcan, mebcan
from iqdbc.car.volkswagen.pq_radar_handler import PQRadarHandler
from iqdbc.car.volkswagen.values import (
CanBus, CarControllerParams, MQB_A0_CARS, VolkswagenFlags, VolkswagenFlagsIQ, apply_pq_stopping_accel,
)
from iqdbc.car.volkswagen.values import CanBus, CarControllerParams, VolkswagenFlags, VolkswagenFlagsIQ
from iqdbc.car.volkswagen.mebutils import LongControlJerk, LongControlLimit, LatControlCurvature
from iqdbc.car.vehicle_model import VehicleModel
from openpilot.system.proprietary_runtime._verified_import import import_verified_module
iq_lvbs_alc = import_verified_module("iqpilot_alc_private", "iqpilot_private.konn3kt.iqlvbs.alc")
iq_lvbs_commander = import_verified_module("iqpilot_commander_private", "iqpilot_private.konn3kt.iqlvbs.iqlvbs_commander")
from openpilot.iqpilot.konn3kt.iqlvbs import alc as iq_lvbs_alc
iqpilot_path = os.path.join(os.path.dirname(__file__), '..', '..', '..')
sys.path.insert(0, iqpilot_path)
@@ -33,19 +27,9 @@ except ImportError:
pass
VisualAlert = structs.CarControl.HUDControl.VisualAlert
AudibleAlert = structs.CarControl.HUDControl.AudibleAlert
LongCtrlState = structs.CarControl.Actuators.LongControlState
def dVisual(CCS, CS):
if CCS == mqbcan:
decelV = CS.tsk_verzoeg_anf
elif CCS == pqcan:
decelV = CS.br8_acc_anf
else:
decelV = False
return decelV
class MQBStandstillManager:
BRAKE_TORQUE_RAMP_RATE = 2000.0 # Nm/s
ASSUMED_WHEEL_RADIUS = 0.328 # m, typical tire rolling radius
@@ -192,6 +176,7 @@ class MQBStandstillManager:
self.prev_accel = accel
return long_active, accel, stopping, starting, esp_starting_override, esp_stopping_override
class CarController(CarControllerBase):
def __init__(self, dbc_names, CP, CP_IQ):
super().__init__(dbc_names, CP, CP_IQ)
@@ -200,11 +185,8 @@ class CarController(CarControllerBase):
self.CAN = CanBus(CP)
self.packer_pt = CANPacker(dbc_names[Bus.pt])
self._pt_tx_bus = self.CAN.pt
if CP.flags & VolkswagenFlags.PQ:
self.CCS = pqcan
if CP.flags & VolkswagenFlagsIQ.IQ_PQ_LOWLINE:
self._pt_tx_bus = self.CAN.aux
elif CP.flags & VolkswagenFlags.MLB:
self.CCS = mlbcan
elif CP.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO):
@@ -229,16 +211,13 @@ class CarController(CarControllerBase):
self.long_jerk_control = LongControlJerk(dt=(DT_CTRL * self.CCP.ACC_CONTROL_STEP)) if self.CP.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO) else None
self.long_limit_control = LongControlLimit(dt=(DT_CTRL * self.CCP.ACC_CONTROL_STEP)) if self.CP.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO) else None
self.gra_acc_counter_last = None
self.motor3_frame_last = None
self.motor3_was_stopping = False
self.motor3_resuming = False
self.sng_handoff_active = False
self.acc_counter_seeded = False
self.klr_counter_last = None
self.eps_timer_soft_disable_alert = False
self.hca_frame_timer_running = 0
self.hca_frame_same_torque = 0
self.accel_last = 0
self.accel_diff = 0
self.long_deviation = 0
self.long_jerklimit = 0
self.HCA_Status = 3
@@ -252,43 +231,19 @@ class CarController(CarControllerBase):
self.radar_disabled_warning_timer = 0
# Check once at init whether the DBC includes MEB_AWV_01 (AEB HUD for radar-disabled camera harness cars)
self._has_aeb_hud_msg = "MEB_AWV_01" in self.packer_pt.dbc.name_to_msg
self.eps_timer_workaround = bool(CP.flags & (VolkswagenFlags.MLB | VolkswagenFlagsIQ.IQ_PQ_TIMEBOMB))
self._pq_patch_checked = not bool(CP.flags & VolkswagenFlags.PQ) or self.eps_timer_workaround
self.eps_timer_workaround = bool(CP.flags & VolkswagenFlags.MLB)
self.hca_frame_timer_resetting = 0
self.hca_frame_low_torque = 0
self.long_override_counter = 0
self.long_disabled_counter = 0
self.standstill_manager = MQBStandstillManager(CP.mass, self.CCP.ACCEL_MIN) if self.CCS == mqbcan else None
self.radar_handler = PQRadarHandler(self.CAN) if self.CCS is pqcan else None
self.blend_stock_radar = False
self.unavailable = False
self.unavailable_hold = 0
self.VM = VehicleModel(CP)
self.is_mqb_a0 = self._is_mqb_a0_car(CP.carFingerprint)
self.LateralController = (
LatControlCurvature(self.CCP.CURVATURE_PID, self.CCP.CURVATURE_LIMITS.CURVATURE_MAX, 1 / (DT_CTRL * self.CCP.STEER_STEP))
if (CP.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO))
else None
)
@staticmethod
def _is_mqb_a0_car(candidate) -> bool:
return candidate in MQB_A0_CARS
def _get_mqb_steering_torque_scale(self, v_ego: float, enabled: bool) -> float:
if enabled and self.CCS == mqbcan:
return float(np.interp(v_ego, [0.4, 3.5, 4.0], [0.8, 0.95, 1.0]))
return 1.0
def _should_spam_mqb_a0_resume(self, CS, enabled: bool) -> bool:
return bool(
enabled and
self.is_mqb_a0 and
self.CCS == mqbcan and
CS.out.standstill and
self.frame % 50 < 15
)
def update(self, CC, CC_IQ, CS, now_nanos):
actuators = CC.actuators
hud_control = CC.hudControl
@@ -297,23 +252,10 @@ class CarController(CarControllerBase):
apply_torque = 0
pqhca5or7Toggle = self._params.get_bool("pqhca5or7Toggle")
eBrakeActive = self._params.get_bool("eBrakeActive")
iq_mqb_steering_lockout = self._params.get_bool("iqMqbSteeringLockout")
iq_mqb_acc_resume = self._params.get_bool("iqMqbAccResume")
self.blend_stock_radar = self._params.get_bool("IQDynamicBlendStockRadar")
if not self._pq_patch_checked:
self._pq_patch_checked = True
self.eps_timer_workaround = not self._params.get_bool("VwPqEpsPatched")
blend_active = bool(self.blend_stock_radar and self.radar_handler is not None and self.CP.openpilotLongitudinalControl)
AngleLateralControl = iq_lvbs_alc.angle_lateral_control_enabled(self, CS)
self.entering = CS.vw_iq_lvbs_alc_entering
self.active = CS.vw_iq_lvbs_alc_active
if hud_control.audibleAlert == AudibleAlert.refuse:
self.unavailable_hold = self.CCP.IQ_PQ_UNAVAILABLE_HUD_FRAMES
else:
self.unavailable_hold = max(0, self.unavailable_hold - 1)
self.unavailable = self.unavailable_hold > 0
CS.force_rhd_for_bsm = getattr(CC, "forceRHDForBSM", False)
CS.enable_predicative_speed_limit = getattr(CC.cruiseControl, "speedLimitPredicative", False)
CS.enable_pred_react_to_speed_limits = getattr(CC.cruiseControl, "speedLimitPredReactToSL", False)
@@ -354,8 +296,7 @@ class CarController(CarControllerBase):
self.steering_power_last = steering_power
else:
if CC.latActive and not AngleLateralControl:
torque_scale = self._get_mqb_steering_torque_scale(CS.out.vEgo, iq_mqb_steering_lockout)
new_torque = int(round(actuators.torque * self.CCP.STEER_MAX * torque_scale))
new_torque = int(round(actuators.torque * self.CCP.STEER_MAX))
apply_torque = apply_driver_steer_torque_limits(new_torque, self.apply_torque_last, CS.out.steeringTorque, self.CCP)
self.hca_frame_timer_running += self.CCP.STEER_STEP
if self.apply_torque_last == apply_torque:
@@ -400,10 +341,10 @@ class CarController(CarControllerBase):
self.eps_timer_soft_disable_alert = self.hca_frame_timer_running > self.CCP.STEER_TIME_ALERT / DT_CTRL
self.apply_torque_last = apply_torque
if not (AngleLateralControl and self.CCS in (mqbcan, pqcan, mlbcan)):
can_sends.append(self.CCS.create_hca_steering_control(self.packer_pt, self._pt_tx_bus, output_torque, self.HCA_Status))
if not (AngleLateralControl and self.CCS in (mqbcan, pqcan)):
can_sends.append(self.CCS.create_hca_steering_control(self.packer_pt, self.CAN.pt, output_torque, self.HCA_Status))
if self.CP.flags & VolkswagenFlags.STOCK_HCA_PRESENT and self.CCS == mqbcan:
if self.CP.flags & VolkswagenFlags.STOCK_HCA_PRESENT:
ea_simulated_torque = float(np.clip(apply_torque * 2, -self.CCP.STEER_MAX, self.CCP.STEER_MAX))
if abs(CS.out.steeringTorque) > abs(ea_simulated_torque):
ea_simulated_torque = CS.out.steeringTorque
@@ -438,7 +379,7 @@ class CarController(CarControllerBase):
if self.frame % self.CCP.ACC_CONTROL_STEP == 0 and self.CP.openpilotLongitudinalControl and not CS.out.radarDisableFailed:
stopping = actuators.longControlState == LongCtrlState.stopping
if self.CP.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO):
starting = actuators.longControlState == LongCtrlState.pid and (CS.esp_hold_confirmation or CS.out.vEgo < 0.25)
starting = actuators.longControlState == LongCtrlState.starting and CS.out.vEgo <= self.CP.vEgoStarting
accel = float(np.clip(actuators.accel, self.CCP.ACCEL_MIN, self.CCP.ACCEL_MAX) if CC.enabled else 0)
long_override = CC.cruiseControl.override or CS.out.gasPressed
@@ -468,7 +409,7 @@ class CarController(CarControllerBase):
self.accel_last = accel
else:
stopping = actuators.longControlState == LongCtrlState.stopping
starting = actuators.longControlState == LongCtrlState.pid and (CS.esp_hold_confirmation or CS.out.vEgo < 0.25)
starting = actuators.longControlState == LongCtrlState.pid and (CS.esp_hold_confirmation or CS.out.vEgo < self.CP.vEgoStopping)
long_active = CC.longActive
accel = actuators.accel
esp_starting_override = None
@@ -483,16 +424,10 @@ class CarController(CarControllerBase):
acc_control = self.CCS.acc_control_value(CS.out.cruiseState.available, long_active, CC.cruiseControl.override, CS.out.accFaulted)
accel = float(np.clip(accel, self.CCP.ACCEL_MIN, self.CCP.ACCEL_MAX) if (long_active or CC.cruiseControl.override) else 0)
self.long_jerklimit = float(interp(accel, [-1.5, -0.3, 0.05], [2.0, 0.52, 0.30]))
self.long_deviation = float(interp(accel, [-0.3, 0.1], [0.0, 0.20]))
self.accel_diff = (0.0019 * (accel - self.accel_last)) + (1 - 0.0019) * self.accel_diff
self.long_jerklimit = 3.0 # AendGrad (0.01 * (np.clip(abs(accel), 0.7, 2))) + (1 - 0.01) * self.long_jerklimit
self.long_deviation = 0.2 # RegelAbw np.interp(abs(accel - self.accel_diff), [0, 0.3, 1.0], [0.02, 0.04, 0.08])
self.accel_last = accel
if blend_active and getattr(CC_IQ, "useRadarAccel", False) and CS.acc_radar_sta_adr == 1 and not self.radar_handler.failed:
accel = float(np.clip(CS.acc_radar_sollbeschl, self.CCP.ACCEL_MIN, self.CCP.ACCEL_MAX))
self.long_jerklimit = CS.acc_radar_aendgrad
self.long_deviation = CS.acc_radar_regelabw
self.accel_last = accel
if self.CCS == mqbcan:
can_sends.extend(self.CCS.create_acc_accel_control(
self.packer_pt, self.CAN.pt, CS.acc_type, accel, acc_control, stopping, starting, CS.esp_hold_confirmation,
@@ -500,19 +435,10 @@ class CarController(CarControllerBase):
esp_starting_override=esp_starting_override, esp_stopping_override=esp_stopping_override,
))
else:
accel = apply_pq_stopping_accel(self.CP.carFingerprint, accel, stopping)
sng_ecd_enabled = bool(self.CP.flags & VolkswagenFlagsIQ.IQ_PQ_SNG_ECD)
low_speed = CS.out.vEgo <= self.CCP.SNG_HANDOFF_SPEED
if sng_ecd_enabled and long_active and low_speed and accel <= 0.05 and not starting:
self.sng_handoff_active = True
else:
self.sng_handoff_active = False
sng_decel_req = float(np.clip(accel, self.CCP.ACCEL_MIN, self.CCP.SNG_HOLD_DECEL_MAX)) if self.sng_handoff_active else 0.0
can_sends.extend(self.CCS.create_acc_accel_control(self.packer_pt, self.CAN.pt, CS.acc_type, accel, acc_control, stopping and not self.motor3_resuming, starting, CS.esp_hold_confirmation, self.long_deviation, self.long_jerklimit, eBrakeActive, sng_active=self.sng_handoff_active))
if sng_ecd_enabled:
can_sends.append(self.CCS.create_sng_handoff_control(self.packer_pt, self.CAN.aux, self.sng_handoff_active, sng_decel_req))
can_sends.extend(self.CCS.create_acc_accel_control(
self.packer_pt, self.CAN.pt, CS.acc_type, accel, acc_control, stopping, starting, CS.esp_hold_confirmation,
self.long_deviation, self.long_jerklimit, eBrakeActive,
))
if (self.CP.flags & VolkswagenFlags.DISABLE_RADAR) and self.CP.openpilotLongitudinalControl and not CS.out.radarDisableFailed:
if self.CP.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO):
@@ -542,16 +468,15 @@ class CarController(CarControllerBase):
can_sends.append(self.CCS.create_lka_hud_control(self.packer_pt, self.CAN.pt, CS.ldw_stock_values, CC.latActive, CS.out.steeringPressed,
hud_alert, hud_control, sound_alert))
else:
can_sends.append(self.CCS.create_lka_hud_control(self.packer_pt, self._pt_tx_bus, CS.ldw_stock_values, CC.latActive, steering_pressed_hud,
can_sends.append(self.CCS.create_lka_hud_control(self.packer_pt, self.CAN.pt, CS.ldw_stock_values, CC.latActive, steering_pressed_hud,
hud_alert, hud_control, self.entering, self.CCS is pqcan and AngleLateralControl, self.active))
if hud_control.leadDistanceBars != self.lead_distance_bars_last:
self.distance_bar_frame = self.frame
if self.frame % self.CCP.ACC_HUD_STEP == 0 and self.CP.openpilotLongitudinalControl:
fcw_alert = hud_control.visualAlert == VisualAlert.fcw
d_unresponsive = hud_control.driverUnresponsive
if self.CP.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO):
fcw_alert = hud_control.visualAlert == VisualAlert.fcw
show_distance_bars = self.frame - self.distance_bar_frame < 400
gap = max(8, CS.out.vEgo * hud_control.leadFollowTime)
distance = max(8, hud_control.leadDistance) if hud_control.leadDistance != 0 else 0
@@ -573,29 +498,23 @@ class CarController(CarControllerBase):
CS.esp_hold_confirmation, distance, gap, fcw_alert, acc_hud_event, speed_limit))
else:
leadDistance = min(8, hud_control.leadDistance) if hud_control.leadDistance != 0 else 0
fcw_alert = hud_control.visualAlert == VisualAlert.fcw
self.leadDistanceBars = min(3, hud_control.leadDistanceBars)
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))
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))
if self.CP.flags & VolkswagenFlags.PQ:
iq_lvbs_commander.update_turn_signals(self, CC, CS, can_sends)
if self.frame % 2 == 0:
self.blinkerActive = CS.leftBlinkerUpdate or CS.rightBlinkerUpdate
leftBlinker = CC.leftBlinker if not self.blinkerActive else False
rightBlinker = CC.rightBlinker if not self.blinkerActive else False
can_sends.append(self.CCS.create_blinker_control(self.packer_pt, self.CAN.pt, leftBlinker, rightBlinker))
if self.CP.openpilotLongitudinalControl and (self.CP.flags & VolkswagenFlags.PQ):
if blend_active:
can_sends.extend(self.radar_handler.update(
self.packer_pt, self.frame, CS,
blend_active=True,
engage_req=getattr(CC_IQ, "radarEngageReq", False),
cancel_req=getattr(CC_IQ, "radarCancelReq", False),
set_speed_kph=getattr(CC_IQ, "radarSetSpeedKph", 0.0),
gap_bars=getattr(CC_IQ, "radarGapBars", 0),
v_ego=CS.out.vEgo,
))
elif self.frame % 2 == 0:
if self.frame % 2 == 0:
can_sends.append(self.CCS.filter_motor2(self.packer_pt, self.CAN.ext, CS.motor2_stock))
can_sends.append(self.CCS.filter_motor5(self.packer_pt, self.CAN.ext, CS.motor5_stock))
gra_send_ready = self.CP.pcmCruise and CS.gra_stock_values["COUNTER"] != self.gra_acc_counter_last
if self.CP.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO):
@@ -607,25 +526,15 @@ class CarController(CarControllerBase):
stock_cancel_pressed = bool(CS.gra_stock_values["GRA_Abbrechen"])
cancel_cmd = stock_cancel_pressed or CC.cruiseControl.cancel
resume_cmd = CC.cruiseControl.resume or self._should_spam_mqb_a0_resume(CS, iq_mqb_acc_resume)
if gra_send_ready and (cancel_cmd or resume_cmd):
if gra_send_ready and (cancel_cmd or CC.cruiseControl.resume):
bus_send = self.CAN.aux if self.CP.flags & VolkswagenFlags.PQ else self.CAN.ext
can_sends.append(self.CCS.create_acc_buttons_control(self.packer_pt, bus_send, CS.gra_stock_values,
cancel=cancel_cmd, resume=resume_cmd))
cancel=cancel_cmd, resume=CC.cruiseControl.resume))
if self.CP.openpilotLongitudinalControl and self.CCS == pqcan and not blend_active:
if self.CP.openpilotLongitudinalControl and self.CCS == pqcan:
if self.frame % 3:
can_sends.append(self.CCS.create_gra_neu(self.packer_pt, self.CAN.ext, CS.gra_stock_values, CC.longActive))
if self.CP.flags & VolkswagenFlagsIQ.IQ_PQ_ACC_FTS_EPB:
is_stopping = actuators.longControlState == LongCtrlState.stopping
if CS.out.vEgo > 0.5 or not CC.longActive:
self.motor3_resuming = False
elif (self.motor3_was_stopping and not is_stopping and self.CP.openpilotLongitudinalControl and CC.longActive):
self.motor3_resuming = True
if self.motor3_resuming and CS.motor3_stock:
can_sends.append(self.CCS.create_motor3_resume(self.packer_pt, self.CAN.aux, CS.motor1_stock, CS.motor3_stock, resume=True))
self.motor3_was_stopping = is_stopping
new_actuators = actuators.as_builder()
new_actuators.torque = self.apply_torque_last / self.CCP.STEER_MAX
@@ -638,6 +547,5 @@ class CarController(CarControllerBase):
self.lead_distance_bars_last = hud_control.leadDistanceBars
self.gra_acc_counter_last = CS.gra_stock_values["COUNTER"]
self.motor3_frame_last = CS.motor3_frame
self.frame += 1
return new_actuators, can_sends

View File

@@ -12,10 +12,7 @@ from iqdbc.car.common.conversions import Conversions as CV
from iqdbc.car.volkswagen.values import CAR, DBC, CanBus, NetworkLocation, RADAR_DISABLE_STATE, TransmissionType, GearShifter, \
CarControllerParams, VolkswagenFlags, VolkswagenFlagsIQ
from iqdbc.car.volkswagen.speed_limit_manager import SpeedLimitManager
from openpilot.system.proprietary_runtime._verified_import import import_verified_module
iq_lvbs_alc = import_verified_module("iqpilot_alc_private", "iqpilot_private.konn3kt.iqlvbs.alc")
vehicle_state = import_verified_module("iqpilot_alc_private", "iqpilot_private.konn3kt.iqlvbs.vehicle_state")
from openpilot.iqpilot.konn3kt.iqlvbs import alc as iq_lvbs_alc
iqpilot_path = os.path.join(os.path.dirname(__file__), '..', '..', '..')
sys.path.insert(0, iqpilot_path)
@@ -48,14 +45,6 @@ class CarState(CarStateBase):
self.ea_hud_stock_values = {}
self.ea_control_stock_values = {}
self.acc_type = 0
self.acc_radar_sollbeschl = 0.0
self.acc_radar_regelabw = 0.0
self.acc_radar_aendgrad = 0.0
self.acc_radar_sta_adr = 0
self.acc_radar_fehler = False
self.acc_radar_v_wunsch = 0.0
self.acc_radar_sta_acc = 0
self.epb_freigabe_ver = False
self.acc_stock_counters: dict[str, int] = {}
self.esp_stopping = False
self.tsk_brake_torque = 0.0
@@ -96,20 +85,7 @@ class CarState(CarStateBase):
self.PQ_ALC_Status_raw = 0
self.alcOverrideAlert = False
self.bremse8_stock = None
self.br8_acc_anf = False
self.tsk_verzoeg_anf = False
self.cruise_main_switch = False
self.motor3_stock = {}
self.motor1_stock = {}
self.motor3_frame = 0
self.motor1_frame = 0
self._odometer_store = vehicle_state.VehicleOdometerStore(CP, self._params)
def _update_odometer(self, ret: structs.CarState, raw_km: float) -> None:
"""Publish the cluster value while proprietary Konn3kt code owns persistence."""
odometer_km = self._odometer_store.record(raw_km)
if odometer_km is not None:
ret.odometer = odometer_km
def _apply_iq_private_flags(self, ret_iq: structs.IQCarState) -> None:
ret_iq.alcOverrideAlert = bool(self.alcOverrideAlert)
@@ -141,9 +117,6 @@ class CarState(CarStateBase):
def _update_mqb_iq_alc_state(self, pt_cp):
iq_lvbs_alc.update_mqb_carstate_alc_state(self, pt_cp)
def _update_mlb_iq_alc_state(self, pt_cp):
iq_lvbs_alc.update_mlb_carstate_alc_state(self, pt_cp)
def _update_pq_iq_alc_state(self, pt_cp):
iq_lvbs_alc.update_pq_carstate_alc_state(self, pt_cp)
@@ -153,8 +126,7 @@ class CarState(CarStateBase):
ext_cp = pt_cp if self.CP.networkLocation == NetworkLocation.fwdCamera else cam_cp
if self.CP.flags & VolkswagenFlags.PQ:
aux_cp = can_parsers.get(Bus.aux)
return self.update_pq(pt_cp, cam_cp, ext_cp, aux_cp)
return self.update_pq(pt_cp, cam_cp, ext_cp)
elif self.CP.flags & VolkswagenFlags.MLB:
br_cp = can_parsers[Bus.aux]
return self.update_mlb(pt_cp, br_cp, cam_cp, ext_cp)
@@ -246,7 +218,6 @@ class CarState(CarStateBase):
self.grade = pt_cp.vl["Motor_16"]["TSK_Steigung"]
acc_limiter_mode = False if cc_only else ext_cp.vl["ACC_02"]["ACC_Gesetzte_Zeitluecke"] == 0
speed_limiter_mode = bool(pt_cp.vl["TSK_06"]["TSK_Limiter_ausgewaehlt"])
self.tsk_verzoeg_anf = bool(pt_cp.vl["TSK_06"]["TSK_Freig_Verzoeg_Anf"])
self._update_mqb_iq_alc_state(pt_cp)
@@ -285,15 +256,12 @@ class CarState(CarStateBase):
ret.buttonEvents = self.create_button_events(pt_cp, self.CCP.BUTTONS)
ret.lowSpeedAlert = self.update_low_speed_alert(ret.vEgo)
allow_lat_only = self._params.get_bool("AllowLateralWhenLongUnavailable") and self._params.get_bool("AolEnabled")
cruise_main_switch = bool(self.gra_stock_values["GRA_Hauptschalter"])
ret.cruiseFaultLateralMode = allow_lat_only and ret.accFaulted and cruise_main_switch
ret.lateralAvailable = ret.cruiseState.available or ret.cruiseFaultLateralMode
ret.blockPcmEnable = ret.cruiseFaultLateralMode
ret.cruiseFaultLateralMode = False
ret.lateralAvailable = ret.cruiseState.available
ret.blockPcmEnable = False
ret.fuelGauge = pt_cp.vl["Kombi_02"]["KBI_Inhalt_Tank"] / 55.0
ret.fuelTankLevelL = pt_cp.vl["Kombi_02"]["KBI_Inhalt_Tank"] # raw liters for konn3kt
self._update_odometer(ret, aux_cp.vl["Kombi_02"]["KBI_Kilometerstand"])
self.cruise_faulted = ret.accFaulted
self._apply_iq_private_flags(ret_iq)
@@ -409,7 +377,6 @@ class CarState(CarStateBase):
psd_06_values = main_cp.vl["PSD_06"] if self.CP.flags & VolkswagenFlags.STOCK_PSD_PRESENT else {}
psd_06_values = pt_cp.vl["PSD_06"] if not psd_06_values and self.CP.flags & VolkswagenFlags.STOCK_PSD_06_PRESENT else psd_06_values
diagnose_01_values = pt_cp.vl["Diagnose_01"] if self.CP.flags & VolkswagenFlags.STOCK_DIAGNOSE_01_PRESENT else {}
self._update_odometer(ret, pt_cp.vl["Diagnose_01"]["KBI_Kilometerstand"])
if self.enable_speed_limit_predicative and not self.enable_predicative_speed_limit:
self.enable_predicative_speed_limit = True
@@ -434,8 +401,10 @@ class CarState(CarStateBase):
ret.espDisabled = bool(pt_cp.vl["ESP_21"]["ESP_Tastung_passiv"])
ret.espActive = bool(pt_cp.vl["ESP_21"]["ESP_Eingriff"])
allow_lat_only = self._params.get_bool("AllowLateralWhenLongUnavailable") and self._params.get_bool("AolEnabled")
cruise_fault_candidate = allow_lat_only and ret.accFaulted and bool(pt_cp.vl["GRA_ACC_01"]["GRA_Hauptschalter"])
allow_lat_only = self._params.get_bool("AllowLateralWhenLongUnavailable")
cruise_main_switch = bool(pt_cp.vl["GRA_ACC_01"]["GRA_Hauptschalter"])
cruise_fault_candidate = allow_lat_only and ret.accFaulted and cruise_main_switch
if cruise_fault_candidate:
self.cruise_faulted_frames += 1
self.cruise_fault_clear_frames = 0
@@ -449,10 +418,12 @@ class CarState(CarStateBase):
self.cruise_fault_lateral_active = False
else:
self.cruise_fault_clear_frames = 0
if not allow_lat_only:
self.cruise_fault_lateral_active = False
self.cruise_faulted_frames = 0
self.cruise_fault_clear_frames = 0
ret.cruiseFaultLateralMode = self.cruise_fault_lateral_active
ret.lateralAvailable = ret.cruiseState.available or ret.cruiseFaultLateralMode
ret.blockPcmEnable = ret.cruiseFaultLateralMode
@@ -482,10 +453,9 @@ class CarState(CarStateBase):
self._apply_iq_private_flags(ret_iq)
return ret, ret_iq
def update_pq(self, pt_cp, cam_cp, ext_cp, aux_cp=None) -> tuple[structs.CarState, structs.IQCarState]:
def update_pq(self, pt_cp, cam_cp, ext_cp) -> tuple[structs.CarState, structs.IQCarState]:
ret = structs.CarState()
ret_iq = structs.IQCarState()
sng_ecd_enabled = bool(self.CP.flags & VolkswagenFlagsIQ.IQ_PQ_SNG_ECD)
# 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"]["BR1_Rad_kmh"] * CV.KPH_TO_MS
@@ -498,13 +468,13 @@ class CarState(CarStateBase):
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
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, ready_confirms_init=False)
ret.steerFaultTemporary, ret.steerFaultPermanent = self.update_hca_state(hca_status)
# Update gas, brakes, and gearshift.
ret.gasPressed = pt_cp.vl["Motor_3"]["MO3_Pedalwert"] > 0
ret.brake = pt_cp.vl["Bremse_5"]["BR5_Bremsdruck"] / 250.0 # FIXME: this is pressure in Bar, not sure what OP expects
ret.brakePressed = bool(pt_cp.vl["Motor_2"]["MO2_BLS"])
ret.parkingBrake = False # bool(pt_cp.vl["Kombi_1"]["Bremsinfo"])
ret.parkingBrake = bool(pt_cp.vl["Kombi_1"]["Bremsinfo"])
# Update gear and/or clutch position data.
if self.CP.transmissionType == TransmissionType.automatic:
@@ -533,11 +503,8 @@ class CarState(CarStateBase):
# Consume blind-spot monitoring info/warning LED states, if available.
# Infostufe: BSM LED on, Warnung: BSM LED flashing
if self.CP.enableBsm:
blindspot_li = bool(ext_cp.vl["SWA_1"]["SWA_Infostufe_SWA_li"]) or bool(ext_cp.vl["SWA_1"]["SWA_Warnung_SWA_li"])
blindspot_re = bool(ext_cp.vl["SWA_1"]["SWA_Infostufe_SWA_re"]) or bool(ext_cp.vl["SWA_1"]["SWA_Warnung_SWA_re"])
force_rhd = self.force_rhd_for_bsm or self._params.get_bool("ForceRHDForBSM")
ret.leftBlindspot = blindspot_re if force_rhd else blindspot_li
ret.rightBlindspot = blindspot_li if force_rhd else blindspot_re
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
@@ -557,18 +524,16 @@ class CarState(CarStateBase):
# Update ACC radar status.
self.acc_type = 0 if cc_only else ext_cp.vl["ACC_System"]["ACS_Typ_ACC"]
cruise_main_switch = bool(pt_cp.vl["Motor_5"]["MO5_GRA_Hauptsch"])
self.cruise_main_switch = cruise_main_switch
cruise_tsk_status = bool(pt_cp.vl["Motor_2"]["MO2_Status_TSK"])
self.cruise_main_switch = cruise_main_switch or cruise_tsk_status
MO2_StaGRA = pt_cp.vl["Motor_2"]["MO2_Sta_GRA"] in (1, 2)
ACS_StaADR = False if cc_only else ext_cp.vl["ACC_System"]["ACS_Sta_ADR"] == 1
cruiseActive = MO2_StaGRA or ACS_StaADR
self.epb_freigabe_ver = bool(aux_cp.vl["EPB_1"]["EP1_Freigabe_Ver"]) if sng_ecd_enabled and not cc_only else False
sng_holding = sng_ecd_enabled and self.epb_freigabe_ver
if cruiseActive or sng_holding:
if cruiseActive:
self.last_cruiseActive = True
elif not MO2_StaGRA and not ACS_StaADR and not sng_holding:
elif not MO2_StaGRA and not ACS_StaADR:
self.last_cruiseActive = False
ret.cruiseState.enabled = self.last_cruiseActive
self.br8_acc_anf = bool(pt_cp.vl["Bremse_8"]["BR8_Sta_ACC_Anf"]) if not cc_only else False
if self.CP.pcmCruise:
if cc_only:
@@ -579,9 +544,9 @@ class CarState(CarStateBase):
cruise_faulted = pt_cp.vl["Motor_2"]["MO2_Sta_GRA"] == 3
ret.accFaulted = cruise_faulted
ret.cruiseState.available = cruise_main_switch and not cruise_faulted
ret.cruiseState.available = (cruise_main_switch or cruise_tsk_status) and not cruise_faulted
allow_lat_only = self._params.get_bool("AllowLateralWhenLongUnavailable") and self._params.get_bool("AolEnabled")
cruise_main_available = cruise_main_switch
cruise_main_available = cruise_main_switch or cruise_tsk_status
ret.cruiseFaultLateralMode = allow_lat_only and cruise_faulted and cruise_main_available
ret.lateralAvailable = ret.cruiseState.available or ret.cruiseFaultLateralMode
ret.blockPcmEnable = ret.cruiseFaultLateralMode
@@ -598,33 +563,6 @@ class CarState(CarStateBase):
ret.cruiseState.speed = 0
self.motor2_stock = pt_cp.vl["Motor_2"]
self.motor5_stock = pt_cp.vl["Motor_5"]
if self.CP.flags & VolkswagenFlagsIQ.IQ_PQ_ACC_FTS_EPB:
self.motor3_stock = aux_cp.vl["Motor_3"]
self.motor1_stock = aux_cp.vl["Motor_1"]
self.motor3_frame += 1
self.motor1_frame += 1
if cc_only:
self.acc_radar_sollbeschl = 0.0
self.acc_radar_regelabw = 0.0
self.acc_radar_aendgrad = 0.0
self.acc_radar_sta_adr = 0
self.acc_radar_fehler = False
self.acc_radar_v_wunsch = 0.0
self.acc_radar_sta_acc = 0
else:
self.acc_radar_sollbeschl = ext_cp.vl["ACC_System"]["ACS_Sollbeschl"]
self.acc_radar_regelabw = ext_cp.vl["ACC_System"]["ACS_zul_Regelabw"]
self.acc_radar_aendgrad = ext_cp.vl["ACC_System"]["ACS_max_AendGrad"]
self.acc_radar_sta_adr = int(ext_cp.vl["ACC_System"]["ACS_Sta_ADR"])
self.acc_radar_fehler = bool(ext_cp.vl["ACC_System"]["ACS_Fehler"])
self.acc_radar_v_wunsch = ext_cp.vl["ACC_GRA_Anzeige"]["ACA_V_Wunsch"]
self.acc_radar_sta_acc = int(ext_cp.vl["ACC_GRA_Anzeige"]["ACA_StaACC"])
ret_iq.accRadarStaAdr = self.acc_radar_sta_adr
ret_iq.accRadarFehler = self.acc_radar_fehler
# 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"],
@@ -639,15 +577,8 @@ class CarState(CarStateBase):
ret.lowSpeedAlert = self.update_low_speed_alert(ret.vEgo)
ret.fuelGauge = 0 # pt_cp.vl["Kombi_1"]["Tankinhalt"] / 55.0
ret.fuelTankLevelL = 0 # pt_cp.vl["Kombi_1"]["Tankinhalt"] # raw liters for konn3kt
if aux_cp is not None:
self._update_odometer(ret, aux_cp.vl["Kombi_3"]["Kilometerstand"])
if self.CP.flags & VolkswagenFlagsIQ.IQ_PQ_ACC_FTS_EPB:
ret.cruiseState.standstill = self.CP.pcmCruise and bool(pt_cp.vl["Bremse_5"]["BR5_Stillstand"]) and ret.cruiseState.enabled
elif sng_ecd_enabled:
ret.cruiseState.standstill = sng_holding and ret.standstill
ret.fuelGauge = pt_cp.vl["Kombi_1"]["Tankinhalt"] / 55.0
ret.fuelTankLevelL = pt_cp.vl["Kombi_1"]["Tankinhalt"] # raw liters for konn3kt
self.cruise_faulted = ret.accFaulted
self.frame += 1
@@ -684,7 +615,6 @@ class CarState(CarStateBase):
ret.accFaulted = pt_cp.vl["TSK_02"]["TSK_Status"] in (3,)
self.parse_mlb_mqb_steering_state(ret, pt_cp)
self._update_mlb_iq_alc_state(pt_cp)
ret.brake = pt_cp.vl["ESP_05"]["ESP_Bremsdruck"] / 250.0
brake_pedal_pressed = bool(pt_cp.vl["Motor_03"]["MO_Fahrer_bremst"])
@@ -718,7 +648,6 @@ class CarState(CarStateBase):
ret.fuelGauge = br_cp.vl["Kombi_02"]["KBI_Inhalt_Tank"] / 55.0
ret.fuelTankLevelL = br_cp.vl["Kombi_02"]["KBI_Inhalt_Tank"] # raw liters for konn3kt
self._update_odometer(ret, br_cp.vl["Kombi_02"]["KBI_Kilometerstand"])
ret.buttonEvents = self.create_button_events(pt_cp, self.CCP.BUTTONS)
@@ -751,11 +680,10 @@ class CarState(CarStateBase):
ret.steerFaultTemporary, ret.steerFaultPermanent = self.update_hca_state(hca_status, drive_mode)
return
def update_hca_state(self, hca_status, drive_mode=True, ready_confirms_init=True):
def update_hca_state(self, hca_status, drive_mode=True):
# Treat 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
init_statuses = ("DISABLED", "READY", "ACTIVE") if ready_confirms_init else ("DISABLED", "ACTIVE")
self.eps_init_complete = self.eps_init_complete or hca_status in init_statuses or self.frame > 1000
self.eps_init_complete = self.eps_init_complete or (hca_status in ("DISABLED", "READY", "ACTIVE") or self.frame > 600)
perm_fault = drive_mode and hca_status == "DISABLED" or (self.eps_init_complete and hca_status == "FAULT")
temp_fault = drive_mode and hca_status in ("REJECTED", "PREEMPTED") or not self.eps_init_complete
return temp_fault, perm_fault
@@ -807,11 +735,8 @@ class CarState(CarStateBase):
]
if CP.flags & VolkswagenFlags.MLB:
pt_messages += [
("Blinkmodi_01", math.nan), # From J519 BCM (is inactive when no lights active, 50Hz when active)
("Kombi_02", math.nan), # Auxiliary-bus cluster odometer
("Blinkmodi_01", math.nan) # From J519 BCM (is inactive when no lights active, 50Hz when active)
]
else:
pt_messages += [("Kombi_02", math.nan)] # Auxiliary-bus cluster odometer
if CP.flags & VolkswagenFlags.STOCK_HCA_PRESENT:
cam_messages += [
("HCA_01", 1), # From R242 Driver assistance camera, 50Hz if steering/1Hz if not
@@ -825,12 +750,8 @@ class CarState(CarStateBase):
@staticmethod
def get_can_parsers_pq(CP):
aux_messages = [("Kombi_3", math.nan)] # Bus 1 cluster odometer
if CP.flags & VolkswagenFlagsIQ.IQ_PQ_ACC_FTS_EPB:
aux_messages.append(("Motor_3", 0))
return {
Bus.pt: CANParser(DBC[CP.carFingerprint][Bus.pt], [], CanBus(CP).powertrain),
Bus.aux: CANParser(DBC[CP.carFingerprint][Bus.pt], aux_messages, CanBus(CP).aux),
Bus.pt: CANParser(DBC[CP.carFingerprint][Bus.pt], [], CanBus(CP).pt),
Bus.cam: CANParser(DBC[CP.carFingerprint][Bus.pt], [], CanBus(CP).cam),
}
@@ -842,7 +763,6 @@ class CarState(CarStateBase):
# TA_01 lives on bus 0 (car ECU / OP-generated when long is active).
# math.nan → ignore_alive=True so it never contributes to can_valid.
("TA_01", math.nan),
("Diagnose_01", math.nan), # Bus 0 cluster odometer
]
if CP.networkLocation == NetworkLocation.fwdCamera:
pt_messages.append(("AWV_03", 1))

View File

@@ -681,14 +681,6 @@ FW_VERSIONS = {
b'\xf1\x873QF907572A \xf1\x890132',
],
},
CAR.VOLKSWAGEN_PASSAT_B7: {
(Ecu.engine, 0x7e0, None): [
b'\xf1\x8703L906018RE\xf1\x899979',
],
(Ecu.fwdCamera, 0x74f, None): [
b'\xf1\x873AA980654D \xf1\x890300\xf1\x82\x0143',
],
},
CAR.VOLKSWAGEN_POLO_MK6: {
(Ecu.engine, 0x7e0, None): [
b'\xf1\x8704C906025H \xf1\x895177',
@@ -724,17 +716,6 @@ FW_VERSIONS = {
b'\xf1\x877N0907572C \xf1\x890211\xf1\x82\x0153',
],
},
CAR.SEAT_ALHAMBRA_MK1: {
(Ecu.engine, 0x7e0, None): [
b'\xf1\x8704L906016HE\xf1\x894635',
],
(Ecu.srs, 0x715, None): [
b'\xf1\x877N0959655D \xf1\x890016\xf1\x82\x0801100705----10--',
],
(Ecu.fwdRadar, 0x757, None): [
b'\xf1\x877N0907572C \xf1\x890211\xf1\x82\x0153',
],
},
CAR.VOLKSWAGEN_TAOS_MK1: {
(Ecu.engine, 0x7e0, None): [
b'\xf1\x8704E906025CK\xf1\x892228',

View File

@@ -1,16 +1,15 @@
import time
from iqdbc.car.common.conversions import Conversions as CV
from iqdbc.car import get_safety_config, structs, uds
from iqdbc.car.carlog import carlog
from iqdbc.car.interfaces import CarInterfaceBase
from iqdbc.car.isotp_parallel_query import IsoTpParallelQuery
from iqdbc.car.volkswagen.carcontroller import CarController
from iqdbc.car.volkswagen.carstate import CarState
from iqdbc.car.volkswagen.values import (
CAR, CanBus, DashcamOnlyReason, NetworkLocation, RADAR_DISABLE_STATE, TransmissionType,
VolkswagenFlags, VolkswagenSafetyFlags, VolkswagenFlagsIQ, get_longitudinal_stopping_speed_override,
)
from iqdbc.car.volkswagen.values import CanBus, CAR, DashcamOnlyReason, NetworkLocation, RADAR_DISABLE_STATE, TransmissionType, VolkswagenFlags, VolkswagenSafetyFlags, VolkswagenFlagsIQ
from iqdbc.car.volkswagen.radar_interface import RadarInterface
from iqdbc.car.common.conversions import Conversions as CV
import sys
import os
iqpilot_path = os.path.join(os.path.dirname(__file__), '..', '..', '..')
@@ -20,11 +19,6 @@ try:
except ImportError:
pass
try:
from openpilot.system.proprietary_runtime._verified_import import import_verified_module
except Exception:
import_verified_module = None
class CarInterface(CarInterfaceBase):
CarState = CarState
@@ -45,16 +39,9 @@ class CarInterface(CarInterfaceBase):
if ret.flags & VolkswagenFlags.PQ:
# Set global PQ35/PQ46/NMS parameters
safety_configs = [get_safety_config(structs.CarParams.SafetyModel.volkswagenPq)]
if candidate == CAR.SEAT_ALHAMBRA_MK1:
ret.flags |= VolkswagenFlagsIQ.IQ_PQ_TIMEBOMB.value
if not (ret.flags & VolkswagenFlagsIQ.IQ_PQ_TIMEBOMB) and _params.get_bool("VwPqEpsPatched"):
ret.minSteerSpeed = 0
if angle_lat_enabled:
ret.flags |= VolkswagenFlagsIQ.IQ_LVBS_ALC_MODULE.value
safety_configs[0].safetyParam |= VolkswagenSafetyFlags.PQ_ALC_MODULE.value
if alpha_long:
ret.flags |= VolkswagenFlagsIQ.IQ_PQ_SNG_ECD.value
safety_configs[0].safetyParam |= VolkswagenSafetyFlags.PQ_SNG_ECD.value
ret.enableBsm = 0x3BA in fingerprint[0] # SWA_1
if 0x440 in fingerprint[0] or docs: # Getriebe_1
@@ -72,13 +59,6 @@ class CarInterface(CarInterfaceBase):
else:
ret.flags |= VolkswagenFlagsIQ.IQ_CC_ONLY_NO_RADAR.value
cc_only_flags = VolkswagenFlagsIQ.IQ_CC_ONLY | VolkswagenFlagsIQ.IQ_CC_ONLY_NO_RADAR
if ret.flags & cc_only_flags:
safety_configs[0].safetyParam |= VolkswagenSafetyFlags.PQ_NO_CAM_BUS.value
if (ret.flags & cc_only_flags) and not fingerprint[0]:
ret.flags |= VolkswagenFlagsIQ.IQ_PQ_LOWLINE.value
safety_configs[0].safetyParam |= VolkswagenSafetyFlags.PQ_LOWLINE.value
if any(msg in fingerprint[1] for msg in (0x1A0, 0xC2)): # Bremse_1, Lenkwinkel_1
ret.networkLocation = NetworkLocation.gateway
else:
@@ -184,20 +164,19 @@ class CarInterface(CarInterfaceBase):
CarInterfaceBase.configure_torque_tune(candidate, ret.lateralTuning)
elif ret.flags & VolkswagenFlags.MLB:
ret.steerActuatorDelay = 0.2
if angle_lat_enabled:
ret.steerControlType = structs.CarParams.SteerControlType.angle
ret.steerAtStandstill = bool(joystick_mode)
else:
CarInterfaceBase.configure_torque_tune(candidate, ret.lateralTuning)
CarInterfaceBase.configure_torque_tune(candidate, ret.lateralTuning)
elif ret.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO):
ret.steerActuatorDelay = 0.3
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]
if angle_lat_enabled:
ret.steerControlType = structs.CarParams.SteerControlType.angle
ret.steerAtStandstill = bool(joystick_mode)
else:
CarInterfaceBase.configure_torque_tune(candidate, ret.lateralTuning)
# Global longitudinal tuning defaults, can be overridden per-vehicle
@@ -221,12 +200,17 @@ class CarInterface(CarInterfaceBase):
if candidate == CAR.PORSCHE_MACAN_MK1:
ret.steerActuatorDelay = 0.07
if candidate == CAR.VOLKSWAGEN_PASSAT_B7 or CAR.SEAT_ALHAMBRA_MK1:
ret.flags |= VolkswagenFlagsIQ.IQ_PQ_ACC_FTS_EPB.value
safety_configs[0].safetyParam |= VolkswagenSafetyFlags.PQ_ACC_FTS_EPB.value
ret.pcmCruise = not ret.openpilotLongitudinalControl
ret.stopAccel = -0.55
if ret.flags & (VolkswagenFlags.MEB | VolkswagenFlags.MQB_EVO):
ret.startingState = True
ret.startAccel = 0.8
ret.vEgoStarting = 0.5
ret.vEgoStopping = 0.1
ret.stopAccel = -0.55
else:
ret.stopAccel = -0.55
ret.vEgoStarting = 0.1
ret.vEgoStopping = 1.5 * CV.KPH_TO_MS if ret.flags & VolkswagenFlags.PQ else 0.1
ret.autoResumeSng = ret.minEnableSpeed == -1
CAN = CanBus(fingerprint=fingerprint)
if CAN.pt >= 4:
@@ -237,18 +221,10 @@ class CarInterface(CarInterfaceBase):
@staticmethod
def pre_init(CP: structs.CarParams, CP_IQ: structs.IQCarParams, can_recv, can_send):
if not (CP.flags & VolkswagenFlags.PQ) or (CP.flags & VolkswagenFlagsIQ.IQ_PQ_TIMEBOMB) or import_verified_module is None:
return
try:
params = Params()
if params.get_bool("VwPqEpsPatched"):
return
flasher = import_verified_module("iqpilot_hephaestusd_private", "iqpilot_private.konn3kt.hephaestus.vw_pq_flasher")
status = flasher.check_eps_patch_status(1, can_recv, can_send)
except Exception:
return
if status == "patched":
params.put_bool("VwPqEpsPatched", True)
# Engine-on check moved to init(): if radar can't be disabled, radarDisableFailed=True
# gates only long control (carcontroller line ~308) while lateral still works.
# Full dashcam mode here was too aggressive — lateral doesn't need radar disabled.
pass
@staticmethod
def init(CP: structs.CarParams, CP_IQ: structs.IQCarParams, can_recv, can_send):
@@ -345,5 +321,4 @@ class CarInterface(CarInterfaceBase):
@staticmethod
def _get_params_iq(stock_cp: structs.CarParams, ret: structs.IQCarParams, candidate, fingerprint: dict[int, dict[int, int]], car_fw: list[structs.CarParams.CarFw], alpha_long: bool, is_release_iq: bool, docs: bool) -> structs.IQCarParams:
ret.longitudinalStoppingSpeedOverride = get_longitudinal_stopping_speed_override(candidate, stock_cp.flags)
return ret

View File

@@ -28,7 +28,7 @@ def create_steering_control(packer, bus, apply_curvature, lkas_enabled, power):
values = {
"Curvature": abs(apply_curvature), # in rad/m
"Curvature_VZ": 1 if apply_curvature > 0 and lkas_enabled else 0,
"Power": power if lkas_enabled else 0,
"Power": 100 if lkas_enabled else 0, # TEST: hard 100%, no ramp
"RequestStatus": 4 if lkas_enabled else 2,
"HighSendRate": lkas_enabled,
}

View File

@@ -270,5 +270,5 @@ class LatControlCurvature():
error = desired_curvature - actual_curvature
freeze_integrator = CC.steerLimited or CS.vEgo < 5
output_curvature = self.pid.update(error, feedforward=desired_curvature_corr, speed=CS.vEgo,
freeze_integrator=freeze_integrator, override=CS.steeringPressed)
freeze_integrator=freeze_integrator, override=False)
return output_curvature

View File

@@ -45,7 +45,7 @@ def create_acc_accel_control(packer, bus, acc_type, accel, acc_control, stopping
return [packer.make_can_msg("ACC_05", bus, values)]
def create_acc_hud_control(packer, bus, acc_hud_status, set_speed, leadDistance, distanceBars, fcw_alert, leadVisible, unavailable, decel, d_unresponsive):
def create_acc_hud_control(packer, bus, acc_hud_status, set_speed, leadDistance, distanceBars, fcw_alert, leadVisible):
values = {}
return packer.make_can_msg("ACC_02", bus, values)

View File

@@ -114,10 +114,10 @@ def create_acc_accel_control(packer, bus, acc_type, accel, acc_control, stopping
"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,
"ACC_zul_Regelabw_oben": 0.2,
"ACC_neg_Sollbeschl_Grad_02": 3.0 if acc_enabled else 0,
"ACC_pos_Sollbeschl_Grad_02": 3.0 if acc_enabled else 0,
"ACC_zul_Regelabw_unten": comfortBand if acc_enabled else 0.2,
"ACC_zul_Regelabw_oben": comfortBand if acc_enabled else 0.2,
"ACC_neg_Sollbeschl_Grad_02": jerkLimit if acc_enabled else 0,
"ACC_pos_Sollbeschl_Grad_02": jerkLimit if acc_enabled else 0,
"ACC_Anfahren": starting if acc_enabled else False,
"ACC_Anhalten": stopping if acc_enabled else False,
}
@@ -149,16 +149,12 @@ def create_acc_accel_control(packer, bus, acc_type, accel, acc_control, stopping
return commands
def create_acc_hud_control(packer, bus, acc_hud_status, set_speed, leadDistance, distanceBars, fcw_alert, leadVisible, unavailable, decel, d_unresponsive):
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
def create_acc_hud_control(packer, bus, acc_hud_status, set_speed, leadDistance, distanceBars, fcw_alert, leadVisible):
values = {
"ACC_Status_Anzeige": acc_hud_status,
"ACC_Wunschgeschw_02": set_speed if set_speed < 250 else 327.36,
"ACC_Gesetzte_Zeitluecke": leadDistanceBars,
"ACC_Optischer_Fahrerhinweis": 1 if fcw_alert else 0,
"ACC_Display_Prio": priodisp,
"ACC_Relevantes_Objekt": leadDistanceBars,
"ACC_Gesetzte_Zeitluecke": distanceBars,
"ACC_Display_Prio": 3,
"ACC_Abstandsindex": leadDistance if leadVisible else 0,
"ACC_Akustik_02": fcw_alert,
}

View File

@@ -1,106 +0,0 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from iqdbc.car.common.conversions import Conversions as CV
from iqdbc.car.volkswagen import pqcan
class PQRadarHandler:
GRA_STEP = 3 # GRA_Neu cadence (~33 Hz), matches stock stalk module
SPOOF_STEP = 2 # Motor_2 / Motor_5 spoof cadence (50 Hz)
ENGAGE_FLOOR = 1.0 * CV.KPH_TO_MS # never SET at/below 1 kph
REENGAGE_FLOOR = 2.0 * CV.KPH_TO_MS # hysteresis: only (re)engage above 2 kph
SETSPEED_TOL_KPH = 1.0 # don't chase set-speed within this band
SHORT_STEP_KPH = 1.0 * CV.MPH_TO_KPH # GRA_*_kurz ~= 1 mph
LONG_STEP_KPH = 5.0 * CV.MPH_TO_KPH # GRA_*_lang ~= 5 mph
TAP_RELEASE_CYCLES = 2 # GRA cycles to release between set-speed taps
def __init__(self, CAN):
self.bus = CAN.ext # radar lives on bus 2 (ext/cam)
self.counter = 0
self.failed = False # ACS_Fehler / irrev_Fehler -> radar dead for the drive
self.want_engaged = False # our belief the radar cruise should be on
self._press_phase = 0 # alternate press/release for SET/cancel discrete edges
self._tap_cooldown = 0 # set-speed tap rate limiter
def reset(self):
self.want_engaged = False
self._press_phase = 0
self._tap_cooldown = 0
@staticmethod
def _map_gap_bars(gap_bars):
if not gap_bars:
return None
return int(min(3, max(1, gap_bars)))
def update(self, packer, frame, CS, *, blend_active, engage_req, cancel_req,
set_speed_kph, gap_bars, v_ego):
can_sends = []
if not blend_active:
self.reset()
return can_sends
if CS.acc_radar_fehler or CS.acc_radar_sta_adr == 3:
self.failed = True
radar_active = (CS.acc_radar_sta_adr == 1) and not self.failed
if self.failed:
self.want_engaged = False
elif cancel_req:
self.want_engaged = False
elif engage_req and v_ego > self.REENGAGE_FLOOR:
self.want_engaged = True
if (frame % self.SPOOF_STEP) == 0:
# Mirror the stock engage handshake ORDER: the radar leads (SET -> ADR active) and only THEN
# does the engine report GRA regulating. Asserting MO2_Sta_GRA=1 before the radar's ADR is
# active presents an impossible state ("engine regulating but ADR didn't initiate it") and the
# radar latches an irreversible fault. So only assert cruise-active once ADR is actually active;
# relay Motor_5 (main switch) as verbatim OEM passthrough throughout.
hold_engaged = self.want_engaged and radar_active
can_sends.append(pqcan.filter_motor2(packer, self.bus, CS.motor2_stock, gra_active=hold_engaged))
can_sends.append(pqcan.filter_motor5(packer, self.bus, CS.motor5_stock, gra_active=False))
if (frame % self.GRA_STEP) == 0:
set_btn = resume_btn = cancel = up_s = down_s = up_l = down_l = False
self._press_phase ^= 1
pressing = self._press_phase == 0
if self.failed:
pass
elif cancel_req:
# Only actually press cancel when the radar is engaged (or overdriven) -- otherwise it's a
# no-op against a passive/off radar and just spams GRA_Abbrechen at ~16 Hz. Fires while
# active and stops once the radar leaves the active state.
cancel = pressing and radar_active
elif self.want_engaged and not radar_active and v_ego > self.ENGAGE_FLOOR:
# Engage with RESUME (GRA_Recall), not SET. Ground-truth from an OEM ACC engage (radar leads:
# RESUME -> ACS_Sta_ADR 2->1 in ~90ms -> engine MO2_Sta_GRA 0->1 ~80ms later) shows this radar
# engages from passive on GRA_Recall; it ignores GRA_Neu_Setzen from that state.
resume_btn = pressing
elif self.want_engaged and radar_active:
if self._tap_cooldown > 0:
self._tap_cooldown -= 1
elif set_speed_kph > 0:
delta = set_speed_kph - CS.acc_radar_v_wunsch
if abs(delta) >= self.SETSPEED_TOL_KPH:
big = abs(delta) >= self.LONG_STEP_KPH
if delta > 0:
up_l, up_s = big, not big
else:
down_l, down_s = big, not big
self._tap_cooldown = self.TAP_RELEASE_CYCLES
self.counter = (self.counter + 1) % 16
can_sends.append(pqcan.create_radar_gra(
packer, self.bus, CS.gra_stock_values, self.counter,
set_btn=set_btn, resume=resume_btn, cancel=cancel, up_short=up_s, down_short=down_s,
up_long=up_l, down_long=down_l, zeitluecke=self._map_gap_bars(gap_bars),
))
return can_sends

View File

@@ -1,7 +1,3 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
def create_hca_steering_control(packer, bus, apply_torque, HCA_Status):
values = {
"LM_Offset": abs(apply_torque),
@@ -99,20 +95,20 @@ def acc_hud_status_value(main_switch_on, acc_faulted, longActive, longOverride):
return hud_status
def create_acc_accel_control(packer, bus, acc_type, accel, acc_control, stopping, starting, esp_hold, comfortBand, jerkLimit, eBrakeActive, sng_active=False):
def create_acc_accel_control(packer, bus, acc_type, accel, acc_control, stopping, starting, esp_hold, comfortBand, jerkLimit, eBrakeActive):
commands = []
acc_enabled = acc_control == 1 and not sng_active
acc_enabled = acc_control == 1
values = {
"ACS_Sta_ADR": 0 if sng_active else acc_control,
"ACS_Sta_ADR": acc_control,
"ACS_StSt_Info": acc_enabled,
"ACS_Typ_ACC": acc_type,
"ACS_Anhaltewunsch": (acc_type == 1 and stopping or eBrakeActive) or sng_active,
"ACS_Anhaltewunsch": acc_type == 1 and stopping or eBrakeActive,
"ACS_FreigSollB": acc_enabled,
"ACS_Sollbeschl": accel if acc_enabled else 3.01,
"ACS_zul_Regelabw": comfortBand if acc_enabled else 1.27,
"ACS_max_AendGrad": jerkLimit if acc_enabled else 5.08,
"ACS_Schubabsch": 0,
"ACS_Schubabsch": 1 if acc_enabled and (accel > 0.05) else 0,
"ACS_MomEingriff": 0,
"ACS_ADR_Schub": 0,
}
@@ -122,14 +118,6 @@ def create_acc_accel_control(packer, bus, acc_type, accel, acc_control, stopping
return commands
def create_sng_handoff_control(packer, bus, handoff_active, decel_req):
values = {
"SNG_HandoffActive": handoff_active,
"SNG_DecelReq": decel_req if handoff_active else 0.0,
}
return packer.make_can_msg("SNG_1", bus, values)
def create_blinker_control(packer, bus, leftBlinker, rightBlinker):
values = {
"BM_rechts": rightBlinker,
@@ -138,71 +126,29 @@ def create_blinker_control(packer, bus, leftBlinker, rightBlinker):
return packer.make_can_msg("Blinkmodi_02", bus, values)
def create_acc_hud_control(packer, bus, acc_hud_status, set_speed, leadDistance, distanceBars, fcw_alert, leadVisible, unavailable, decel, d_unresponsive):
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
def create_acc_hud_control(packer, bus, acc_hud_status, set_speed, leadDistance, distanceBars, fcw_alert, leadVisible):
if distanceBars == 1:
leadDistanceBars = 2
elif distanceBars == 2:
leadDistanceBars = 3
elif distanceBars == 3:
leadDistanceBars = 4
else:
leadDistanceBars = 2
values = {
"ACA_StaACC": acc_hud_status,
"ACA_AnzDisplay": 1 if acc_hud_status in (3, 4) else 0,
"ACA_Zeitluecke": leadDistanceBars,
"ACA_V_Wunsch": set_speed,
"ACA_gemZeitl": min(15, max(1, int(round(leadDistance)))) if leadVisible else 0,
"ACA_PrioDisp": priodisp,
"ACA_Akustik1": d_unresponsive,
# "ACA_Fahrerhinw": unavailable,
"ACA_PrioDisp": 3,
"ACA_Akustik2": fcw_alert,
"ACA_ACC_Verz": decel,
}
return packer.make_can_msg("ACC_GRA_Anzeige", bus, values)
def filter_motor2(packer, bus, motor2_stock, gra_active=False):
values = dict(motor2_stock)
if gra_active:
values.update({
"MO2_Sta_GRA": 1,
"MO2_Status_TSK": 1,
})
else:
values.update({
"MO2_Sta_GRA": 0,
})
return packer.make_can_msg("Motor_2", bus, values)
def filter_motor5(packer, bus, motor5_stock, gra_active=False):
values = dict(motor5_stock)
if gra_active:
values["MO5_GRA_Hauptsch"] = 1
return packer.make_can_msg("Motor_5", bus, values)
def create_motor3_resume(packer, bus, motor1_stock, motor3_stock, resume=False):
values = dict(motor3_stock)
values_motor1 = dict(motor1_stock)
if resume:
values["MO3_Pedalwert"] = values_motor1["MO1_Pedalwert"]
return packer.make_can_msg("Motor_3", bus, values)
def create_radar_gra(packer, bus, gra_stock, counter, set_btn=False, cancel=False, resume=False,
up_short=False, down_short=False, up_long=False, down_long=False, zeitluecke=None):
values = {s: gra_stock[s] for s in [
"GRA_Hauptschalt", # ACC main switch passthrough
"GRA_Typ_Hauptschalt", # momentary vs latching
"GRA_Kodierinfo", # configuration
"GRA_Sender", # CAN originator
]}
def filter_motor2(packer, bus, motor2_stock):
values = motor2_stock
values.update({
"COUNTER": counter % 16,
"GRA_Neu_Setzen": 1 if set_btn else 0,
"GRA_Abbrechen": 1 if cancel else 0,
"GRA_Recall": 1 if resume else 0,
"GRA_Up_kurz": 1 if up_short else 0,
"GRA_Down_kurz": 1 if down_short else 0,
"GRA_Up_lang": 1 if up_long else 0,
"GRA_Down_lang": 1 if down_long else 0,
"MO2_Sta_GRA": 0,
})
if zeitluecke is not None:
values["GRA_Zeitluecke"] = zeitluecke
return packer.make_can_msg("GRA_Neu", bus, values)
return packer.make_can_msg("Motor_2", bus, values)

View File

@@ -1,49 +0,0 @@
from types import SimpleNamespace
from iqdbc.car.volkswagen import mqbcan, pqcan
from iqdbc.car.volkswagen.carcontroller import CarController
from iqdbc.car.volkswagen.values import CAR, MQB_A0_CARS
def test_is_mqb_a0_car_matches_expected_platforms():
assert CAR.VOLKSWAGEN_POLO_MK6 in MQB_A0_CARS
assert CAR.VOLKSWAGEN_TCROSS_MK1 in MQB_A0_CARS
assert CAR.SKODA_FABIA_MK4 in MQB_A0_CARS
assert CAR.SKODA_KAMIQ_MK1 in MQB_A0_CARS
assert CarController._is_mqb_a0_car(CAR.VOLKSWAGEN_POLO_MK6)
assert CarController._is_mqb_a0_car(CAR.VOLKSWAGEN_TCROSS_MK1)
assert CarController._is_mqb_a0_car(CAR.SKODA_FABIA_MK4)
assert CarController._is_mqb_a0_car(CAR.SKODA_KAMIQ_MK1)
assert not CarController._is_mqb_a0_car(CAR.VOLKSWAGEN_GOLF_MK7)
def test_mqb_steering_torque_scale_only_changes_when_toggle_enabled():
controller = object.__new__(CarController)
controller.CCS = mqbcan
assert controller._get_mqb_steering_torque_scale(0.4, False) == 1.0
assert controller._get_mqb_steering_torque_scale(0.4, True) == 0.8
assert controller._get_mqb_steering_torque_scale(4.0, True) == 1.0
controller.CCS = pqcan
assert controller._get_mqb_steering_torque_scale(0.4, True) == 1.0
def test_mqb_a0_resume_spam_requires_toggle_platform_and_window():
controller = object.__new__(CarController)
controller.CCS = mqbcan
controller.is_mqb_a0 = True
controller.frame = 10
cs = SimpleNamespace(out=SimpleNamespace(standstill=True))
assert controller._should_spam_mqb_a0_resume(cs, True)
controller.frame = 20
assert not controller._should_spam_mqb_a0_resume(cs, True)
controller.frame = 10
controller.is_mqb_a0 = False
assert not controller._should_spam_mqb_a0_resume(cs, True)
controller.is_mqb_a0 = True
assert not controller._should_spam_mqb_a0_resume(cs, False)

View File

@@ -1,186 +0,0 @@
"""
Regression guard for the class of bug fixed in carstate.py's Diagnose_1 (cluster clock,
removed) and EPB_1 (stop-and-go hold, now gated on PQ_SNG_ECD): reading a CAN message via
`some_cp.vl["MsgName"]["Signal"]` without declaring it in get_can_parsers_pq() lazily adds
it via VLDict.__getitem__ -> CANParser._add_message(key, freq=None), which is NOT the same
as ignore-alive (that's what math.nan is for -- see
iqdbc/can/tests/test_packer_parser.py::test_lazy_add_not_ignore_alive). An undeclared
message defaults to "assume ~1Hz, must be seen within ~10s", so if the real car never sends
it, CarState.canValid gets stuck False forever.
Rather than fuzzing every VolkswagenFlagsIQ combination (most are unreachable through
interface.py's real detection logic), each fixture below is the CarParams captured from a
real konn3kt route for a real car of that variant. We run CarState.update() once (no CAN
data needs to be fed -- .vl[...] lazily adds and returns default-zero values regardless of
whether the parser has ever seen a real frame) while recording every message name accessed,
then assert that set is a subset of what that real car actually transmits (per its captured
CAN fingerprint). A message accessed but not in the real fingerprint is exactly the bug
class this guards against.
"""
from iqdbc.can.parser import VLDict
from iqdbc.car import structs
from iqdbc.car.volkswagen.carstate import CarState
from iqdbc.car.volkswagen.values import CAR, Bus
class _RecordingVLDict(VLDict):
"""Records every message name read via .vl[...], including ones lazily added for
messages never declared in get_can_parsers_pq's message list. Applied by reclassing
a live CANParser.vl instance in place (rather than replacing it with a fresh object)
so any messages already registered at CANParser construction time, and the parser
back-reference _add_message needs, are preserved."""
accessed: set[str]
def __getitem__(self, key):
if isinstance(key, str):
self.accessed.add(key)
return super().__getitem__(key)
def _make_car_params(car_fingerprint, flags, network_location, transmission_type, enable_bsm, pcm_cruise):
CP = structs.CarParams.new_message()
CP.carFingerprint = car_fingerprint
CP.flags = flags
CP.radarUnavailable = True
CP.networkLocation = network_location
CP.transmissionType = transmission_type
CP.enableBsm = enable_bsm
CP.pcmCruise = pcm_cruise
CP.minSteerSpeed = 0.0
CP_IQ = structs.IQCarParams()
return CP, CP_IQ
# Real per-variant CarParams captured from konn3kt routes (not hand-derived), so each
# fixture reflects an actual car rather than a guess at which flag combos are reachable.
# known_bus1/known_bus2 are the exact message names present in that car's real CAN
# fingerprint. Bus.pt and Bus.aux both listen on physical bus 1 for PQ (CanBus.powertrain
# == CanBus.aux == 1); Bus.cam listens on physical bus 2 (CanBus.cam == 2) -- see CanBus in
# iqdbc/car/volkswagen/values.py. UNK_* (unrecognized DBC addresses) are dropped.
FIXTURES = [
dict(
name="jetta_mk6_base",
route="a0d99b85ff06857b|0000003a--135c88414f",
car_fingerprint=CAR.VOLKSWAGEN_JETTA_MK6,
flags=2,
network_location="gateway",
transmission_type="automatic",
enable_bsm=False,
pcm_cruise=True,
known_bus1={
'ACC_GRA_Anzeige', 'ACC_System', 'AWV', 'Airbag_1', 'Airbag_2', 'BSG_Last', 'Bremse_1',
'Bremse_10', 'Bremse_11', 'Bremse_2', 'Bremse_3', 'Bremse_4', 'Bremse_5', 'Bremse_8',
'Bremse_9', 'Diagnose_1', 'Einheiten_1', 'GRA_Neu', 'Gate_Komf_1', 'Gate_Komf_2',
'Getriebe_1', 'Getriebe_2', 'Getriebe_4', 'Ident', 'Klima_1', 'Kombi_1', 'Kombi_2',
'Kombi_3', 'Lenkhilfe_1', 'Lenkhilfe_2', 'Lenkhilfe_3', 'Lenkwinkel_1', 'Motor_1',
'Motor_10', 'Motor_12', 'Motor_2', 'Motor_3', 'Motor_5', 'Motor_6', 'Motor_7', 'Motor_8',
'Motor_Bremse', 'Motor_Flexia', 'Soll_Verbauliste_neu', 'Systeminfo_1', 'Waehlhebel_1',
},
known_bus2={
'ACC_GRA_Anzeige', 'ACC_System', 'AWV', 'Airbag_1', 'BSG_Last', 'Bremse_1', 'Bremse_10',
'Bremse_11', 'Bremse_2', 'Bremse_3', 'Bremse_5', 'Bremse_8', 'Diagnose_1', 'Einheiten_1',
'GRA_Neu', 'Gate_Komf_1', 'Gate_Komf_2', 'Getriebe_1', 'Ident', 'Kombi_1', 'Kombi_2',
'Kombi_3', 'Lenkhilfe_2', 'Lenkhilfe_3', 'Lenkwinkel_1', 'Motor_1', 'Motor_10', 'Motor_2',
'Motor_3', 'Motor_5', 'Motor_Flexia', 'Soll_Verbauliste_neu', 'Systeminfo_1',
},
),
dict(
name="passat_nms_sng_ecd",
route="0f53129ed44f6920|00000031--0c1aea511e",
car_fingerprint=CAR.VOLKSWAGEN_PASSAT_NMS,
flags=524418, # PQ | IQ_LVBS_ALC_MODULE | IQ_PQ_SNG_ECD
network_location="gateway",
transmission_type="automatic",
enable_bsm=False,
pcm_cruise=False,
known_bus1={
'ACC_GRA_Anzeige', 'ACC_System', 'AWV', 'Airbag_1', 'Airbag_2', 'BSG_Last', 'Bremse_1',
'Bremse_10', 'Bremse_11', 'Bremse_2', 'Bremse_3', 'Bremse_4', 'Bremse_5', 'Bremse_8',
'Bremse_9', 'Diagnose_1', 'EPB_1', 'EPB_2', 'Einheiten_1', 'GRA_Neu', 'Gate_Komf_1',
'Gate_Komf_2', 'Getriebe_1', 'Getriebe_2', 'Ident', 'Klima_1', 'Kombi_1', 'Kombi_2',
'Kombi_3', 'Lenkhilfe_1', 'Lenkhilfe_2', 'Lenkhilfe_3', 'Lenkwinkel_1', 'Motor_1',
'Motor_10', 'Motor_12', 'Motor_13', 'Motor_2', 'Motor_3', 'Motor_5', 'Motor_6', 'Motor_7',
'Motor_8', 'Motor_Bremse', 'Motor_Flexia', 'Soll_Verbauliste_neu', 'Systeminfo_1',
},
known_bus2={
'ACC_GRA_Anzeige', 'ACC_System', 'AWV', 'Airbag_1', 'BSG_Last', 'Bremse_1', 'Bremse_10',
'Bremse_11', 'Bremse_2', 'Bremse_3', 'Bremse_5', 'Bremse_8', 'Diagnose_1', 'EPB_1',
'Einheiten_1', 'GRA_Neu', 'Gate_Komf_1', 'Gate_Komf_2', 'Getriebe_1', 'Ident', 'Kombi_1',
'Kombi_2', 'Kombi_3', 'Lenkhilfe_2', 'Lenkhilfe_3', 'Lenkwinkel_1', 'Motor_1', 'Motor_10',
'Motor_2', 'Motor_3', 'Motor_5', 'Motor_Flexia', 'Soll_Verbauliste_neu', 'Systeminfo_1',
},
),
dict(
name="passat_b7_acc_fts_epb",
route="20e3cd4f0d5f39d1|0000008e--274d762bac",
car_fingerprint=CAR.VOLKSWAGEN_PASSAT_B7,
flags=262146, # PQ | IQ_PQ_ACC_FTS_EPB
network_location="gateway",
transmission_type="automatic",
enable_bsm=True,
pcm_cruise=False,
known_bus1={
'ACC_GRA_Anzeige', 'ACC_System', 'AWV', 'Airbag_1', 'Airbag_2', 'BSG_Last', 'Bremse_1',
'Bremse_10', 'Bremse_11', 'Bremse_2', 'Bremse_3', 'Bremse_4', 'Bremse_5', 'Bremse_8',
'Bremse_9', 'Daempfer_1', 'Diagnose_1', 'EPB_1', 'Einheiten_1', 'GRA_Neu', 'Gate_Komf_1',
'Gate_Komf_2', 'Getriebe_1', 'Getriebe_2', 'Getriebe_4', 'HCA_1', 'Ident', 'Klima_1',
'Kombi_1', 'Kombi_2', 'Kombi_3', 'Lenkhilfe_1', 'Lenkhilfe_2', 'Lenkhilfe_3',
'Lenkwinkel_1', 'Motor_1', 'Motor_10', 'Motor_12', 'Motor_2', 'Motor_3', 'Motor_5',
'Motor_6', 'Motor_7', 'Motor_8', 'Motor_Bremse', 'Motor_Flexia', 'Parkhilfe_01',
'Soll_Verbauliste_neu', 'Systeminfo_1', 'Waehlhebel_1',
},
known_bus2={
'ACC_GRA_Anzeige', 'ACC_System', 'AWV', 'Airbag_1', 'BSG_Last', 'Bremse_1', 'Bremse_10',
'Bremse_11', 'Bremse_2', 'Bremse_3', 'Bremse_5', 'Bremse_8', 'Diagnose_1', 'EPB_1',
'Einheiten_1', 'GRA_Neu', 'Gate_Komf_1', 'Gate_Komf_2', 'Getriebe_1', 'HCA_1', 'Ident',
'Kombi_1', 'Kombi_2', 'Kombi_3', 'LDW_Status', 'Lenkhilfe_2', 'Lenkhilfe_3',
'Lenkwinkel_1', 'Motor_1', 'Motor_10', 'Motor_2', 'Motor_3', 'Motor_5', 'Motor_Flexia',
'Parkhilfe_01', 'RDK_Status', 'SWA_1', 'Soll_Verbauliste_neu', 'Systeminfo_1',
},
),
]
class TestVolkswagenPqCanValid:
def test_only_reads_messages_the_real_car_sends(self, subtests):
for fixture in FIXTURES:
with subtests.test(car=fixture["name"]):
CP, CP_IQ = _make_car_params(
fixture["car_fingerprint"], fixture["flags"], fixture["network_location"],
fixture["transmission_type"], fixture["enable_bsm"], fixture["pcm_cruise"],
)
CS = CarState(CP, CP_IQ)
can_parsers = CS.get_can_parsers(CP, CP_IQ)
recorders = {}
for bus, parser in can_parsers.items():
if parser is None:
continue
parser.vl.__class__ = _RecordingVLDict
parser.vl.accessed = set()
recorders[bus] = parser.vl
# no CAN data is fed: .vl[...] lazily adds and returns default-zero values
# regardless of whether the parser has ever seen a real frame, so this alone
# is enough to harvest every message name update_pq() touches for this variant
CS.update(can_parsers)
accessed_bus1 = recorders[Bus.pt].accessed | recorders[Bus.aux].accessed
accessed_bus2 = recorders[Bus.cam].accessed
extra_bus1 = accessed_bus1 - fixture["known_bus1"]
extra_bus2 = accessed_bus2 - fixture["known_bus2"]
assert not extra_bus1, (
f"{fixture['name']}: code reads {sorted(extra_bus1)} on bus 1 (pt/aux) but the "
f"real car (route {fixture['route']}) never transmits it -- this needs to be "
f"gated behind whatever flag makes it optional, or declared with math.nan if it's "
f"genuinely on-demand/rare, not read unconditionally"
)
assert not extra_bus2, (
f"{fixture['name']}: code reads {sorted(extra_bus2)} on bus 2 (cam) but the "
f"real car (route {fixture['route']}) never transmits it -- this needs to be "
f"gated behind whatever flag makes it optional, or declared with math.nan if it's "
f"genuinely on-demand/rare, not read unconditionally"
)

View File

@@ -1,22 +0,0 @@
import pytest
from iqdbc.car.volkswagen.values import (
CAR, PASSAT_B7_STOP_ACCEL, PASSAT_B7_STOPPING_SPEED, PQ_STOPPING_SPEED,
VolkswagenFlags, apply_pq_stopping_accel, get_longitudinal_stopping_speed_override,
)
@pytest.mark.parametrize("candidate, flags, expected", [
(CAR.VOLKSWAGEN_PASSAT_B7, VolkswagenFlags.PQ, PASSAT_B7_STOPPING_SPEED),
(CAR.VOLKSWAGEN_JETTA_MK6, VolkswagenFlags.PQ, PQ_STOPPING_SPEED),
(CAR.VOLKSWAGEN_GOLF_MK7, 0, 0.0),
(CAR.VOLKSWAGEN_ID4_MK1, VolkswagenFlags.MEB, 0.0),
])
def test_stopping_speed_override(candidate, flags, expected):
assert get_longitudinal_stopping_speed_override(candidate, flags) == expected
def test_passat_b7_stop_accel_is_exact():
assert apply_pq_stopping_accel(CAR.VOLKSWAGEN_PASSAT_B7, -0.2, True) == PASSAT_B7_STOP_ACCEL == -0.55
assert apply_pq_stopping_accel(CAR.VOLKSWAGEN_PASSAT_B7, -0.2, False) == -0.2
assert apply_pq_stopping_accel(CAR.VOLKSWAGEN_JETTA_MK6, -0.2, True) == -0.2

View File

@@ -43,10 +43,6 @@ class TestVolkswagenPlatformConfigs:
if len(shared_chassis_codes) == 0:
continue
# A shared chassis code is unambiguous when the VIN WMI separates the candidates.
if platform.config.wmis.isdisjoint(comp.config.wmis):
continue
platform_model_years = getattr(platform.config, "model_years", set())
comp_model_years = getattr(comp.config, "model_years", set())
if platform_model_years and comp_model_years and platform_model_years.isdisjoint(comp_model_years):
@@ -55,13 +51,10 @@ class TestVolkswagenPlatformConfigs:
assert set() == shared_chassis_codes, f"Shared chassis codes: {comp}"
def test_custom_fuzzy_fingerprinting(self, subtests):
all_radar_fw = list({fw for ecus in FW_VERSIONS.values() for fw in ecus.get((Ecu.fwdRadar, 0x757, None), [])})
all_radar_fw = list({fw for ecus in FW_VERSIONS.values() for fw in ecus[Ecu.fwdRadar, 0x757, None]})
for platform in CAR:
with subtests.test(platform=platform.name):
# Fuzzy matching keys off the radar ECU (CHECK_FUZZY_ECUS), so a platform that
# declares no radar ECU can never be VIN-matched and should never be expected to match.
platform_has_radar = (Ecu.fwdRadar, 0x757, None) in FW_VERSIONS.get(platform, {})
for wmi in WMI:
for chassis_code in platform.config.chassis_codes | {"00"}:
platform_model_years = getattr(platform.config, "model_years", set())
@@ -77,7 +70,7 @@ class TestVolkswagenPlatformConfigs:
for radar_fw in random.sample(all_radar_fw, 5) + [b'\xf1\x875Q0907572G \xf1\x890571', b'\xf1\x877H9907572AA\xf1\x890396']:
model_year_match = len(platform_model_years) == 0 or model_year in platform_model_years
should_match = ((wmi in platform.config.wmis and chassis_code in platform.config.chassis_codes and model_year_match) and
radar_fw in all_radar_fw and platform_has_radar)
radar_fw in all_radar_fw)
live_fws = {(0x757, None): [radar_fw]}
matches = FW_QUERY_CONFIG.match_fw_to_car_fuzzy(live_fws, vin, FW_VERSIONS)
@@ -113,17 +106,3 @@ def test_button_enable_recovers_once_cruise_fault_clears():
)]
assert state.update_button_enable(button_events)
def test_pq_hca_ready_does_not_complete_eps_initialization():
state = object.__new__(CarState)
state.eps_init_complete = False
state.frame = 0
assert state.update_hca_state("READY", ready_confirms_init=False) == (True, False)
state.frame = 317
assert state.update_hca_state("FAULT", ready_confirms_init=False) == (True, False)
state.frame = 1001
assert state.update_hca_state("FAULT", ready_confirms_init=False) == (False, True)

View File

@@ -35,10 +35,6 @@ class CanBus(CanBusBase):
# NetworkLocation.gateway: powertrain CAN
return 1
@property
def powertrain(self) -> int:
return 1
@property
def main(self) -> int:
return 1
@@ -53,11 +49,17 @@ class CanBus(CanBusBase):
# ADAS / Extended CAN, side of the relay with the ACC radar
return 2
@property
def eps(self) -> int:
return 4
@property
def car(self) -> int:
return 6
# Extra Tolerances For Road Variance
AVERAGE_ROAD_ROLL = 0.06
PQ_STOPPING_SPEED = 1.5 * CV.KPH_TO_MS
PASSAT_B7_STOPPING_SPEED = 0.55 * CV.KPH_TO_MS
PASSAT_B7_STOP_ACCEL = -0.55
class CarControllerParams:
@@ -82,8 +84,6 @@ class CarControllerParams:
AEB_CONTROL_STEP = 2 # ACC_10 frequency 50Hz
AEB_HUD_STEP = 20 # ACC_15 frequency 5Hz
VW_LOW_SPEED_STATE_SPEED = 0.5 # m/s; below this, always force starting or stopping in MQB legacy long
SNG_HANDOFF_SPEED = 5.0 * CV.KPH_TO_MS
SNG_HOLD_DECEL_MAX = -0.1
# 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
@@ -99,8 +99,7 @@ class CarControllerParams:
STEER_LOW_TORQUE = int(STEER_MAX * 0.20) # Steer timer mitigation performed when torque output under 20%
STEER_TIME_LOW_TORQUE = 0.5 # Wait for this duration of STEER_LOW_TORQUE to begin mitigation
STEER_TIME_STUCK_TORQUE = 1.9 # EPS limits same torque to 6 seconds, reset timer 3x within that period
STEER_TIME_RESET = 1.1 # Duration of HCA disable needed for effective EPS timer reset'
IQ_PQ_UNAVAILABLE_HUD_FRAMES = 25
STEER_TIME_RESET = 1.1 # Duration of HCA disable needed for effective EPS timer reset
DEFAULT_MIN_STEER_SPEED = 0.4 # m/s, newer EPS racks fault below this speed, don't show a low speed alert
@@ -116,8 +115,8 @@ class CarControllerParams:
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 = 150 # Max HCA reached in 0.60s (STEER_MAX / (50Hz * 0.60))
self.STEER_DELTA_DOWN = 300 # Min HCA reached in 0.60s (STEER_MAX / (50Hz * 0.60))
self.STEER_DELTA_UP = 10 # Max HCA reached in 0.60s (STEER_MAX / (50Hz * 0.60))
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"]["GE1_Wahl_Pos"]
@@ -285,10 +284,6 @@ class VolkswagenSafetyFlags(IntFlag):
ALLOW_LONG_ACCEL_WITH_GAS_PRESSED = 8
DISABLE_RADAR = 16
PQ_ALC_MODULE = 32
PQ_LOWLINE = 64
PQ_NO_CAM_BUS = 128
PQ_ACC_FTS_EPB = 256
PQ_SNG_ECD = 512
class VolkswagenFlags(IntFlag):
@@ -316,10 +311,6 @@ class VolkswagenFlagsIQ(IntFlag):
IQ_CC_ONLY = 1 << 5 # CC only mode with radar (has AEB)
IQ_CC_ONLY_NO_RADAR = 1 << 6 # CC only mode without radar
IQ_LVBS_ALC_MODULE = 1 << 7 # IQ.Lvbs VW ALC hardware module present / intended active path
IQ_PQ_LOWLINE = 1 << 17 # Non-ECAN lateral-only PQ: bus 0 dead, TX on bus 1 (ptCAN)
IQ_PQ_ACC_FTS_EPB = 1 << 18 # B7 TRW450: ACC FtS + EPB hold, Motor_1 resume spoof on bus 1
IQ_PQ_SNG_ECD = 1 << 19
IQ_PQ_TIMEBOMB = 1 << 20
RADAR_DISABLE_STATE = {"error": False}
@@ -342,7 +333,6 @@ class VolkswagenMQBPlatformConfig(PlatformConfig):
# on camera-integrated cars, as we lose too many ECUs to reliably identify the vehicle
chassis_codes: set[str] = field(default_factory=set)
wmis: set[WMI] = field(default_factory=set)
model_years: set[str] = field(default_factory=set)
@dataclass
@@ -552,35 +542,24 @@ class CAR(Platforms):
VolkswagenCarSpecs(mass=1551, wheelbase=2.79),
chassis_codes={"3C", "3G"},
wmis={WMI.VOLKSWAGEN_EUROPE_CAR},
model_years={"F", "G", "H", "J", "K", "L", "M", "N"}, # 2015-2022
)
VOLKSWAGEN_PASSAT_MK7 = VolkswagenPQPlatformConfig(
[VWCarDocs("Volkswagen Passat 2.0 TDI 2014")],
VolkswagenCarSpecs(mass=1836, wheelbase=2.70, steerRatio=13.0, minSteerSpeed=31 * CV.KPH_TO_MS),
chassis_codes={"3C"},
wmis={WMI.VOLKSWAGEN_EUROPE_CAR},
model_years={"E"}, # 2014
)
VOLKSWAGEN_PASSAT_NMS = VolkswagenPQPlatformConfig(
[VWCarDocs("Volkswagen Passat NMS 2015-17")],
VolkswagenCarSpecs(mass=1503, wheelbase=2.80),
chassis_codes={"A3"},
wmis={WMI.VOLKSWAGEN_USA_CAR},
# NMS and NMS+ share chassis code A3; disambiguate by model year
model_years={"F", "G", "H"}, # 2015-2017
)
VOLKSWAGEN_PASSAT_NMS_PLUS = VolkswagenPQPlatformConfig(
[VWCarDocs("Volkswagen Passat NMS 2018-22")],
VolkswagenCarSpecs(mass=1503, wheelbase=2.80, minEnableSpeed=20 * CV.KPH_TO_MS),
chassis_codes={"A3"},
wmis={WMI.VOLKSWAGEN_USA_CAR},
model_years={"J", "K", "L", "M", "N"}, # 2018-2022
)
VOLKSWAGEN_PASSAT_B7 = VolkswagenPQPlatformConfig(
[VWCarDocs("Volkswagen Passat B7 2008-2011")],
VolkswagenCarSpecs(mass=1503, wheelbase=2.712, steerRatio=16.4, minSteerSpeed=0),
chassis_codes={"A3"},
wmis={WMI.VOLKSWAGEN_USA_CAR},
)
VOLKSWAGEN_POLO_MK6 = VolkswagenMQBPlatformConfig(
[
@@ -594,19 +573,12 @@ class CAR(Platforms):
VOLKSWAGEN_SHARAN_MK2 = VolkswagenPQPlatformConfig(
[
VWCarDocs("Volkswagen Sharan 2018-22"),
VWCarDocs("SEAT Alhambra 2018-20"),
],
VolkswagenCarSpecs(mass=1639, wheelbase=2.92),
chassis_codes={"7N"},
wmis={WMI.VOLKSWAGEN_EUROPE_CAR},
)
SEAT_ALHAMBRA_MK1 = VolkswagenPQPlatformConfig(
[
VWCarDocs("SEAT Alhambra 2018-20"),
],
VolkswagenCarSpecs(mass=1639, wheelbase=2.92, minSteerSpeed=50 * CV.KPH_TO_MS),
chassis_codes={"7N"},
wmis={WMI.SEAT},
)
VOLKSWAGEN_TAOS_MK1 = VolkswagenMQBPlatformConfig(
[VWCarDocs("Volkswagen Taos 2022-24")],
VolkswagenCarSpecs(mass=1498, wheelbase=2.69),
@@ -908,24 +880,4 @@ FW_QUERY_CONFIG = FwQueryConfig(
match_fw_to_car_fuzzy=match_fw_to_car_fuzzy,
)
MQB_A0_CARS = {
CAR.VOLKSWAGEN_POLO_MK6,
CAR.VOLKSWAGEN_TCROSS_MK1,
CAR.SKODA_FABIA_MK4,
CAR.SKODA_KAMIQ_MK1,
}
def get_longitudinal_stopping_speed_override(candidate: CAR, flags: int) -> float:
if candidate == CAR.VOLKSWAGEN_PASSAT_B7:
return PASSAT_B7_STOPPING_SPEED
if flags & VolkswagenFlags.PQ:
return PQ_STOPPING_SPEED
return 0.0
def apply_pq_stopping_accel(candidate: CAR, accel: float, stopping: bool) -> float:
return PASSAT_B7_STOP_ACCEL if candidate == CAR.VOLKSWAGEN_PASSAT_B7 and stopping else accel
DBC = CAR.create_dbc_map()