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

IQ.Pilot Release Commit @ d400490

Browse Source
This commit is contained in:
IQ.Lvbs CI [bot]
2026-07-25 14:42:10 -05:00
parent f0ef157344
commit 729bde8fb9
66 changed files with 435 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
iqdbc_repo/conftest.py
View File

@@ -4,6 +4,59 @@ collect_ignore_glob = [
"iqdbc/safety/tests/misra/cppcheck/",
]
# --- host test harness -------------------------------------------------------
# tesla/volkswagen carstate+carcontroller import the closed-source ALC / odometer
# runtime via import_verified_module(), which gates on a rootfs integrity manifest
# that only exists on-device. Off-device (host CI / dev), install minimal stubs so
# the whole car interface set is importable and test_car_interfaces can collect and
# run. On-device the manifest is present and these stubs are never installed, so the
# real verified modules are used.
import os as _os
if not _os.path.exists("/usr/libexec/iqpilot/runtime_integrity.json"):
import sys as _sys
import types as _types
def _stub_module(name, attrs):
mod = _types.ModuleType(name)
for key, value in attrs.items():
setattr(mod, key, value)
_sys.modules[name] = mod
_noop = lambda *args, **kwargs: None # noqa: E731
class _VehicleOdometerStoreStub:
def __init__(self, *args, **kwargs):
pass
def record(self, km):
return km
_stub_module("iqpilot_private.konn3kt.iqlvbs.vehicle_state",
{"VehicleOdometerStore": _VehicleOdometerStoreStub})
_stub_module("iqpilot_private.konn3kt.iqlvbs.alc", {
"angle_lateral_control_enabled": lambda *a, **k: False,
"update_vw_alc": _noop,
"append_private_apd": _noop,
"update_mqb_carstate_alc_state": _noop,
"update_mlb_carstate_alc_state": _noop,
"update_pq_carstate_alc_state": _noop,
})
_stub_module("iqpilot_private.konn3kt.iqlvbs.iqlvbs_commander", {})
_stub_module("iqpilot_private.konn3kt.hephaestus.vw_pq_flasher", {})
try:
from openpilot.system.proprietary_runtime import _verified_import as _vi
_real_import_verified_module = _vi.import_verified_module
def _import_verified_module(bundle, module):
return _sys.modules.get(module) or _real_import_verified_module(bundle, module)
_vi.import_verified_module = _import_verified_module
except Exception:
pass
# --- end host test harness ---------------------------------------------------
_ABSTRACT_BASES = frozenset(("CarSafetyTest", "AolSafetyTestBase", "SafetyTest", "SafetyTestBase"))
def _method_from_base(cls, method_name):

9
iqdbc_repo/iqdbc/car/chrysler/carstate.py
View File

@@ -4,8 +4,8 @@ from iqdbc.car.chrysler.values import DBC, STEER_THRESHOLD, RAM_CARS
from iqdbc.car.common.conversions import Conversions as CV
from iqdbc.car.interfaces import CarStateBase
from iqdbc.iqpilot.car.chrysler.carstate_ext import CarStateExt
from iqdbc.iqpilot.car.chrysler.aol import AolCarState
from iqdbc.iqpilot.car.chrysler.carstate_ext import CarStateExt
ButtonType = structs.CarState.ButtonEvent.Type
@@ -114,7 +114,10 @@ class CarState(CarStateBase, AolCarState, CarStateExt):
@staticmethod
def get_can_parsers(CP, CP_IQ):
pt_messages: list = []
cam_messages: list = []
AolCarState.get_parser(CP, pt_messages, cam_messages)
return {
Bus.pt: CANParser(DBC[CP.carFingerprint][Bus.pt], [], 0),
Bus.cam: CANParser(DBC[CP.carFingerprint][Bus.pt], [], 2),
Bus.pt: CANParser(DBC[CP.carFingerprint][Bus.pt], pt_messages, 0),
Bus.cam: CANParser(DBC[CP.carFingerprint][Bus.pt], cam_messages, 2),
}

4
iqdbc_repo/iqdbc/car/chrysler/fingerprints.py
View File

@@ -1,10 +1,10 @@
""" AUTO-FORMATTED USING iqdbc/car/debug/format_fingerprints.py, EDIT STRUCTURE THERE."""
from iqdbc.car.structs import CarParams
from iqdbc.car.chrysler.values import CAR
from iqdbc.iqpilot.car.fingerprints_ext import merge_fw_versions
from iqdbc.iqpilot.car.chrysler.fingerprints_ext import FW_VERSIONS_EXT
Ecu = CarParams.Ecu
FW_VERSIONS = {
@@ -789,4 +789,6 @@ FW_VERSIONS = {
},
}
FW_VERSIONS = merge_fw_versions(FW_VERSIONS, FW_VERSIONS_EXT)

3
iqdbc_repo/iqdbc/car/chrysler/interface.py
View File

@@ -100,9 +100,10 @@ class CarInterface(CarInterfaceBase):
stock_cp.wheelbase = 3.79
stock_cp.steerRatio = 19.
# LKAS heartbeat on bus 0 (msg 0x4FF) means the camera is on the ADAS bus and
# IQ.Pilot can steer down to a standstill.
if 0x4FF in fingerprint[0]:
ret.flags |= ChryslerFlagsIQ.NO_MIN_STEERING_SPEED.value
stock_cp.minSteerSpeed = 0.
return ret

7
iqdbc_repo/iqdbc/car/ford/carstate.py
View File

@@ -5,17 +5,14 @@ from iqdbc.car.ford.fordcan import CanBus
from iqdbc.car.ford.values import DBC, CarControllerParams, FordFlags
from iqdbc.car.interfaces import CarStateBase
from iqdbc.iqpilot.car.ford.aol import AolCarState
ButtonType = structs.CarState.ButtonEvent.Type
GearShifter = structs.CarState.GearShifter
TransmissionType = structs.CarParams.TransmissionType
class CarState(CarStateBase, AolCarState):
class CarState(CarStateBase):
def __init__(self, CP, CP_IQ):
CarStateBase.__init__(self, CP, CP_IQ)
AolCarState.__init__(self, CP, CP_IQ)
can_define = CANDefine(DBC[CP.carFingerprint][Bus.pt])
if CP.transmissionType == TransmissionType.automatic:
self.shifter_values = can_define.dv["PowertrainData_10"]["TrnRng_D_Rq"]
@@ -112,8 +109,6 @@ class CarState(CarStateBase, AolCarState):
self.acc_tja_status_stock_values = cp_cam.vl["ACCDATA_3"]
self.lkas_status_stock_values = cp_cam.vl["IPMA_Data"]
AolCarState.update_aol(self, ret, can_parsers)
ret.buttonEvents = [
*create_button_events(self.distance_button, prev_distance_button, {1: ButtonType.gapAdjustCruise}),
*create_button_events(self.lc_button, prev_lc_button, {1: ButtonType.lkas}),

5
iqdbc_repo/iqdbc/car/gm/fingerprints.py
View File

@@ -1,7 +1,6 @@
# ruff: noqa: E501
""" AUTO-FORMATTED USING iqdbc/car/debug/format_fingerprints.py, EDIT STRUCTURE THERE."""
from iqdbc.car.gm.values import CAR
from iqdbc.iqpilot.car.fingerprints_ext import merge_fingerprints
from iqdbc.iqpilot.car.gm.fingerprints_ext import FINGERPRINTS_EXT
@@ -54,6 +53,9 @@ FINGERPRINTS = {
}],
CAR.CHEVROLET_SILVERADO: [{
190: 6, 193: 8, 197: 8, 201: 8, 208: 8, 209: 7, 211: 2, 241: 6, 249: 8, 257: 8, 288: 5, 289: 8, 298: 8, 304: 3, 309: 8, 311: 8, 313: 8, 320: 4, 322: 7, 328: 1, 352: 5, 381: 8, 384: 4, 386: 8, 388: 8, 413: 8, 451: 8, 452: 8, 453: 6, 455: 7, 460: 5, 463: 3, 479: 3, 481: 7, 485: 8, 489: 8, 497: 8, 500: 6, 501: 8, 528: 5, 532: 6, 534: 2, 560: 8, 562: 8, 563: 5, 565: 5, 587: 8, 608: 8, 609: 6, 610: 6, 611: 6, 612: 8, 613: 8, 707: 8, 715: 8, 717: 5, 761: 7, 789: 5, 800: 6, 801: 8, 810: 8, 840: 5, 842: 5, 844: 8, 848: 4, 869: 4, 880: 6, 977: 8, 1001: 8, 1011: 6, 1017: 8, 1020: 8, 1033: 7, 1034: 7, 1217: 8, 1221: 5, 1233: 8, 1249: 8, 1259: 8, 1261: 7, 1263: 4, 1265: 8, 1267: 1, 1271: 8, 1280: 4, 1296: 4, 1300: 8, 1611: 8, 1930: 7
},
{
190: 6, 193: 8, 197: 8, 201: 8, 208: 8, 209: 7, 211: 2, 241: 6, 249: 8, 257: 3, 288: 5, 289: 8, 298: 8, 304: 3, 309: 8, 311: 8, 313: 8, 320: 4, 322: 7, 328: 1, 352: 5, 381: 8, 384: 4, 386: 8, 388: 8, 413: 8, 451: 8, 452: 8, 453: 6, 455: 7, 460: 5, 463: 3, 479: 3, 481: 7, 485: 8, 489: 8, 497: 8, 500: 6, 501: 8, 528: 5, 532: 6, 534: 2, 560: 8, 562: 8, 563: 5, 565: 5, 587: 8, 608: 8, 609: 6, 610: 6, 611: 6, 612: 8, 613: 8, 707: 8, 715: 8, 717: 5, 761: 7, 789: 5, 800: 6, 801: 8, 810: 8, 840: 5, 842: 5, 844: 8, 848: 4, 869: 4, 880: 6, 977: 8, 1001: 8, 1011: 6, 1017: 8, 1020: 8, 1033: 7, 1034: 7, 1217: 8, 1221: 5, 1233: 8, 1249: 8, 1259: 8, 1261: 7, 1263: 4, 1265: 8, 1267: 1, 1271: 8, 1280: 4, 1296: 4, 1300: 8, 1611: 8, 1930: 7
}],
CAR.CHEVROLET_EQUINOX: [{
190: 6, 193: 8, 197: 8, 201: 8, 209: 7, 211: 2, 241: 6, 249: 8, 257: 8, 288: 5, 289: 8, 298: 8, 304: 1, 309: 8, 311: 8, 313: 8, 320: 3, 328: 1, 352: 5, 381: 8, 384: 4, 386: 8, 388: 8, 413: 8, 451: 8, 452: 8, 453: 6, 455: 7, 463: 3, 479: 3, 481: 7, 485: 8, 489: 8, 497: 8, 500: 6, 501: 8, 510: 8, 528: 5, 532: 6, 560: 8, 562: 8, 563: 5, 565: 5, 587: 8, 608: 8, 609: 6, 610: 6, 611: 6, 612: 8, 613: 8, 707: 8, 715: 8, 717: 5, 753: 5, 761: 7, 789: 5, 800: 6, 810: 8, 840: 5, 842: 5, 844: 8, 869: 4, 880: 6, 977: 8, 1001: 8, 1011: 6, 1017: 8, 1020: 8, 1033: 7, 1034: 7, 1217: 8, 1221: 5, 1233: 8, 1249: 8, 1259: 8, 1261: 7, 1263: 4, 1265: 8, 1267: 1, 1271: 8, 1280: 4, 1296: 4, 1300: 8, 1611: 8, 1930: 7
@@ -78,4 +80,5 @@ FINGERPRINTS = {
FW_VERSIONS: dict[str, dict[tuple, list[bytes]]] = {
}
FINGERPRINTS = merge_fingerprints(FINGERPRINTS, FINGERPRINTS_EXT)

15
iqdbc_repo/iqdbc/car/gm/interface.py
View File

@@ -16,7 +16,7 @@ from iqdbc.iqpilot.car.gm.values_ext import GMFlagsIQ, GMSafetyFlagsIQ
TransmissionType = structs.CarParams.TransmissionType
NetworkLocation = structs.CarParams.NetworkLocation
# iqpilot-specific torque parameters for Bolt cars that actually use the d parameter
# Bolt Non-ACC uses the 4th (d) tune parameter; stock tunes zero it out.
NON_LINEAR_TORQUE_PARAMS_IQ = {
CAR.CHEVROLET_BOLT_NON_ACC: [2.24, 1.1, 0.28, -0.07],
CAR.CHEVROLET_BOLT_NON_ACC_1ST_GEN: [1.8, 1.1, 0.3, -0.045],
@@ -241,11 +241,10 @@ class CarInterface(CarInterfaceBase, CarInterfaceExt):
CAR.CHEVROLET_BOLT_NON_ACC_2ND_GEN, CAR.CHEVROLET_TRAILBLAZER_NON_ACC_2ND_GEN):
stock_cp.steerActuatorDelay = 0.2
CarInterfaceBase.configure_torque_tune(candidate, stock_cp.lateralTuning)
elif candidate in (CAR.CHEVROLET_EQUINOX_NON_ACC_3RD_GEN, ):
CarInterfaceBase.configure_torque_tune(candidate, stock_cp.lateralTuning)
# NON_ACC vehicles should use camera car speed thresholds
# Non-ACC cars steer/long via the forward camera and pcmCruise, not the ASCM.
if ret.flags & GMFlagsIQ.NON_ACC:
stock_cp.dashcamOnly = False
stock_cp.alphaLongitudinalAvailable = False
@@ -254,13 +253,13 @@ class CarInterface(CarInterfaceBase, CarInterfaceExt):
stock_cp.pcmCruise = True
stock_cp.safetyConfigs[0].safetyParam |= GMSafetyFlags.HW_CAM.value
ret.iqSafetyFlags |= GMSafetyFlagsIQ.NON_ACC
stock_cp.minEnableSpeed = 24 * CV.MPH_TO_MS # 24 mph
stock_cp.minSteerSpeed = 3.0 # ~6 mph
stock_cp.minEnableSpeed = 24 * CV.MPH_TO_MS
stock_cp.minSteerSpeed = 3.0
# dashcamOnly platforms: untested platforms, need user validations
# Untested Non-ACC platforms ship dashcam-only pending user validation.
if candidate in (CAR.CHEVROLET_BOLT_NON_ACC_2ND_GEN, CAR.CHEVROLET_EQUINOX_NON_ACC_3RD_GEN,
CAR.CHEVROLET_SUBURBAN_NON_ACC_11TH_GEN, CAR.CADILLAC_CT6_NON_ACC_1ST_GEN, CAR.CHEVROLET_TRAILBLAZER_NON_ACC_2ND_GEN,
CAR.CADILLAC_XT5_NON_ACC_1ST_GEN):
CAR.CHEVROLET_SUBURBAN_NON_ACC_11TH_GEN, CAR.CADILLAC_CT6_NON_ACC_1ST_GEN,
CAR.CHEVROLET_TRAILBLAZER_NON_ACC_2ND_GEN, CAR.CADILLAC_XT5_NON_ACC_1ST_GEN):
stock_cp.dashcamOnly = True
return ret

27
iqdbc_repo/iqdbc/car/gm/values.py
View File

@@ -4,9 +4,8 @@ from enum import Enum, IntFlag
from iqdbc.car import Bus, PlatformConfig, DbcDict, Platforms, CarSpecs
from iqdbc.car.structs import CarParams
from iqdbc.car.docs_definitions import CarDocs, CarFootnote, CarHarness, CarParts, Column, SupportType
from iqdbc.car.fw_query_definitions import FwQueryConfig, Request, StdQueries
from iqdbc.iqpilot.car.gm.values_ext import GMFlagsIQ
from iqdbc.car.fw_query_definitions import FwQueryConfig, Request, StdQueries
Ecu = CarParams.Ecu
@@ -88,13 +87,6 @@ class GMCarDocs(CarDocs):
self.car_parts = CarParts.common([CarHarness.obd_ii])
@dataclass
class GMNonAccCarDocs(GMCarDocs):
package: str = "No Adaptive Cruise Control (Non-ACC)"
support_type: SupportType = SupportType.COMMUNITY
support_link: str = "community"
@dataclass(frozen=True, kw_only=True)
class GMCarSpecs(CarSpecs):
tireStiffnessFactor: float = 0.444 # not optimized yet
@@ -123,6 +115,13 @@ class GMSDGMPlatformConfig(GMPlatformConfig):
self.car_docs = []
@dataclass
class GMNonAccCarDocs(GMCarDocs):
package: str = "No Adaptive Cruise Control (Non-ACC)"
support_type: SupportType = SupportType.COMMUNITY
support_link: str = "community"
@dataclass
class GMNonSccPlatformConfig(GMPlatformConfig):
def init(self):
@@ -209,13 +208,7 @@ class CAR(Platforms):
GMCarSpecs(mass=2490, wheelbase=2.94, steerRatio=17.3, centerToFrontRatio=0.5, tireStiffnessFactor=1.0),
)
# port extensions
# Separate car def is required when there is no ASCM
# (for now) unless there is a way to detect it when it has been unplugged...
# CHEVROLET_VOLT_CC = GMNonSccPlatformConfig(
# [GMNonAccCarDocs("Chevrolet Volt LT 2017-18")],
# CHEVROLET_VOLT.specs,
# )
# IQ.Pilot Non-ACC camera-harness ports (no factory adaptive cruise).
CHEVROLET_BOLT_NON_ACC = GMNonSccPlatformConfig(
[GMNonAccCarDocs("Chevrolet Bolt EV Non-ACC 2017")],
CHEVROLET_BOLT_EUV.specs,
@@ -333,7 +326,7 @@ FW_QUERY_CONFIG = FwQueryConfig(
# TODO: detect most of these sets live
EV_CAR = {CAR.CHEVROLET_VOLT, CAR.CHEVROLET_VOLT_2019, CAR.CHEVROLET_BOLT_EUV,
# port extensions, Non-ACC
# Non-ACC EV ports
CAR.CHEVROLET_BOLT_NON_ACC, CAR.CHEVROLET_BOLT_NON_ACC_1ST_GEN, CAR.CHEVROLET_BOLT_NON_ACC_2ND_GEN}
# We're integrated at the camera with VOACC on these cars (instead of ASCM w/ OBD-II harness)

1
iqdbc_repo/iqdbc/car/nissan/carstate.py
View File

@@ -5,6 +5,7 @@ from iqdbc.car import Bus, create_button_events, structs
from iqdbc.car.common.conversions import Conversions as CV
from iqdbc.car.interfaces import CarStateBase
from iqdbc.car.nissan.values import CAR, DBC, CarControllerParams
from iqdbc.iqpilot.car.nissan.carstate_ext import CarStateExt
ButtonType = structs.CarState.ButtonEvent.Type

2
iqdbc_repo/iqdbc/car/rivian/interface.py
View File

@@ -39,6 +39,8 @@ 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:
# Longitudinal harness upgrade advertises msg 0x31a on bus 5; when present it
# unlocks radar, blind-spot and openpilot longitudinal on Rivian.
if 0x31a in fingerprint[5]:
ret.flags |= RivianFlagsIQ.LONGITUDINAL_HARNESS_UPGRADE.value
stock_cp.radarUnavailable = False

11
iqdbc_repo/iqdbc/car/torque_data/substitute.toml
View File

@@ -99,3 +99,14 @@ legend = ["LAT_ACCEL_FACTOR", "MAX_LAT_ACCEL_MEASURED", "FRICTION"]
"SUBARU_FORESTER_PREGLOBAL" = "SUBARU_IMPREZA"
"SUBARU_LEGACY_PREGLOBAL" = "SUBARU_IMPREZA"
"SUBARU_ASCENT" = "SUBARU_FORESTER"
# IQ.Pilot GM Non-ACC ports borrow their base platform's lateral tune
"CHEVROLET_BOLT_NON_ACC" = "CHEVROLET_BOLT_EUV"
"CHEVROLET_BOLT_NON_ACC_1ST_GEN" = "CHEVROLET_BOLT_EUV"
"CHEVROLET_BOLT_NON_ACC_2ND_GEN" = "CHEVROLET_BOLT_EUV"
"CHEVROLET_EQUINOX_NON_ACC_3RD_GEN" = "CHEVROLET_EQUINOX"
"CHEVROLET_SUBURBAN_NON_ACC_11TH_GEN" = "CHEVROLET_SILVERADO"
"CADILLAC_CT6_NON_ACC_1ST_GEN" = "CHEVROLET_VOLT"
"CHEVROLET_TRAILBLAZER_NON_ACC_2ND_GEN" = "CHEVROLET_TRAILBLAZER"
"CHEVROLET_MALIBU_NON_ACC_9TH_GEN" = "CHEVROLET_VOLT"
"CADILLAC_XT5_NON_ACC_1ST_GEN" = "CADILLAC_XT4"

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

@@ -171,7 +171,7 @@ class CarInterface(CarInterfaceBase):
ret.steerLimitTimer = 0.4
if ret.flags & VolkswagenFlags.PQ:
ret.steerActuatorDelay = 0.2
ret.longitudinalTuning.kfDEPRECATED = 1.2
ret.longitudinalTuning.kf = 1.2
ret.longitudinalTuning.kpBP = [0.]
ret.longitudinalTuning.kpV = [.45]
ret.longitudinalTuning.kiBP = [0.]

11
iqdbc_repo/iqdbc/car/volkswagen/tests/test_stopping.py
View File

@@ -1,5 +1,6 @@
import pytest
from iqdbc.car.volkswagen.interface import CarInterface
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,
@@ -20,3 +21,13 @@ 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
@pytest.mark.parametrize("candidate", [
CAR.VOLKSWAGEN_JETTA_MK6,
CAR.VOLKSWAGEN_PASSAT_NMS,
])
def test_pq_longitudinal_feedforward_uses_active_schema_field(candidate):
fingerprint = {bus: {} for bus in range(8)}
cp = CarInterface.get_params(candidate, fingerprint, [], alpha_long=True, is_release=False, docs=False)
assert cp.longitudinalTuning.kf == pytest.approx(1.2)

3
iqdbc_repo/iqdbc/iqpilot/car/chrysler/__init__.py
View File

@@ -1,3 +0,0 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""

52
iqdbc_repo/iqdbc/iqpilot/car/chrysler/aol.py
View File

@@ -1,21 +1,23 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""
Always-on-Lateral adapter for Chrysler. CarState reads the LKAS toggle button
(and the forwarded LKAS heartbeat); CarController tracks the AOL state and, when
AOL is available, drives the LKAS_DISABLED bit the dash reads from the heartbeat.
"""
from enum import StrEnum
from collections import namedtuple
from iqdbc.car import Bus, structs
from iqdbc.car.chrysler.values import RAM_CARS
from iqdbc.iqpilot.aol_base import AolCarStateBase
from iqdbc.can.parser import CANParser
AolDataIQ = namedtuple("AolDataIQ",
["enable_aol", "paused", "lkas_disabled"])
AolDataIQ = namedtuple("AolDataIQ", ["enable_aol", "paused", "lkas_disabled"])
ButtonType = structs.CarState.ButtonEvent.Type
_HEARTBIT_FIELDS = ("LKAS_DISABLED", "AUTO_HIGH_BEAM", "FORWARD_1", "FORWARD_2", "FORWARD_3")
class AolCarController:
def __init__(self):
@@ -23,28 +25,18 @@ class AolCarController:
@staticmethod
def create_lkas_heartbit(packer, lkas_heartbit, aol):
# LKAS_HEARTBIT (0x2D9) LKAS heartbeat
values = {s: lkas_heartbit[s] for s in [
"LKAS_DISABLED",
"AUTO_HIGH_BEAM",
"FORWARD_1",
"FORWARD_2",
"FORWARD_3",
]}
values = {name: lkas_heartbit[name] for name in _HEARTBIT_FIELDS}
if aol.enable_aol:
values["LKAS_DISABLED"] = 1 if aol.lkas_disabled else 0
return packer.make_can_msg("LKAS_HEARTBIT", 0, values)
@staticmethod
def aol_status_update(CC: structs.CarControl, CC_IQ: structs.IQCarControl, CS) -> AolDataIQ:
enable_aol = CC_IQ.aol.available
paused = CC_IQ.aol.enabled and not CC.latActive
# A tap of the LKAS button flips the driver's "LKAS disabled" preference.
if any(be.type == ButtonType.lkas and be.pressed for be in CS.out.buttonEvents):
CS.lkas_disabled = not CS.lkas_disabled
return AolDataIQ(enable_aol, paused, CS.lkas_disabled)
def update(self, CC: structs.CarControl, CC_IQ: structs.IQCarControl, CS) -> None:
@@ -55,35 +47,27 @@ class AolCarState(AolCarStateBase):
def __init__(self, CP: structs.CarParams, CP_IQ: structs.IQCarParams):
super().__init__(CP, CP_IQ)
self.lkas_heartbit = 0
self.init_lkas_disabled = False
self.lkas_disabled = False
@staticmethod
def get_parser(CP, pt_messages, cam_messages) -> None:
if CP.carFingerprint in RAM_CARS:
pt_messages += [
("Center_Stack_2", 1),
]
pt_messages.append(("Center_Stack_2", 1))
else:
pt_messages.append(("TRACTION_BUTTON", 1))
cam_messages.append(("LKAS_HEARTBIT", 1))
def get_lkas_button(self, cp, cp_cam):
def _read_lkas_button(self, cp, cp_cam) -> int:
if self.CP.carFingerprint in RAM_CARS:
lkas_button = cp.vl["Center_Stack_2"]["LKAS_Button"]
else:
lkas_button = cp.vl["TRACTION_BUTTON"]["TOGGLE_LKAS"]
self.lkas_heartbit = cp_cam.vl["LKAS_HEARTBIT"]
if not self.init_lkas_disabled:
self.lkas_disabled = cp_cam.vl["LKAS_HEARTBIT"]["LKAS_DISABLED"]
self.init_lkas_disabled = True
return cp.vl["Center_Stack_2"]["LKAS_Button"]
return lkas_button
self.lkas_heartbit = cp_cam.vl["LKAS_HEARTBIT"]
if not self.init_lkas_disabled:
self.lkas_disabled = cp_cam.vl["LKAS_HEARTBIT"]["LKAS_DISABLED"]
self.init_lkas_disabled = True
return cp.vl["TRACTION_BUTTON"]["TOGGLE_LKAS"]
def update_aol(self, ret: structs.CarState, can_parsers: dict[StrEnum, CANParser]) -> None:
cp = can_parsers[Bus.pt]
cp_cam = can_parsers[Bus.cam]
self.prev_lkas_button = self.lkas_button
self.lkas_button = self.get_lkas_button(cp, cp_cam)
self.lkas_button = self._read_lkas_button(can_parsers[Bus.pt], can_parsers[Bus.cam])

8
iqdbc_repo/iqdbc/iqpilot/car/chrysler/carcontroller_ext.py
View File

@@ -1,5 +1,8 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
Chrysler low-speed steering gate: overrides the stock LKAS control bit when the
no-min-steering-speed option is set, and holds the RAM DT engagement window.
"""
from iqdbc.car import structs
from iqdbc.car.interfaces import CarStateBase
@@ -16,8 +19,9 @@ class CarControllerExt:
def get_lkas_control_bit(self, CS: CarStateBase, CC: structs.CarControl, lkas_control_bit: bool) -> bool:
if self.CP_IQ.flags & ChryslerFlagsIQ.NO_MIN_STEERING_SPEED:
lkas_control_bit = CC.latActive
elif self.CP.carFingerprint in RAM_DT:
return CC.latActive
if self.CP.carFingerprint in RAM_DT:
if self.CP.minEnableSpeed <= CS.out.vEgo <= self.CP.minEnableSpeed + 0.5:
lkas_control_bit = True
if self.CP.minEnableSpeed >= 14.5 and CS.out.gearShifter != GearShifter.drive:

29
iqdbc_repo/iqdbc/iqpilot/car/chrysler/carstate_ext.py
View File

@@ -1,13 +1,13 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""
Chrysler cruise-button reader: emits edge ButtonEvents for the ACC steering-wheel
buttons so IQ.Pilot can react to accel/decel/cancel/resume.
"""
from enum import StrEnum
from iqdbc.car import Bus, structs
from iqdbc.can.parser import CANParser
from iqdbc.car.chrysler.values import RAM_HD
from iqdbc.iqpilot.car.chrysler.values_ext import BUTTONS
@@ -15,23 +15,18 @@ class CarStateExt:
def __init__(self, CP, CP_IQ):
self.CP = CP
self.CP_IQ = CP_IQ
self.button_events = []
self.button_events: list = []
self.button_states = {button.event_type: False for button in BUTTONS}
def update(self, ret: structs.CarState, ret_iq: structs.IQCarState, can_parsers: dict[StrEnum, CANParser]):
def update(self, ret: structs.CarState, ret_iq: structs.IQCarState, can_parsers: dict[StrEnum, CANParser]) -> None:
cp = can_parsers[Bus.pt]
button_events = []
events = []
for button in BUTTONS:
state = (cp.vl[button.can_addr][button.can_msg] in button.values)
if self.button_states[button.event_type] != state:
pressed = cp.vl[button.can_addr][button.can_msg] in button.values
if pressed != self.button_states[button.event_type]:
event = structs.CarState.ButtonEvent.new_message()
event.type = button.event_type
event.pressed = state
button_events.append(event)
self.button_states[button.event_type] = state
self.button_events = button_events
if self.CP.carFingerprint in RAM_HD:
ret.steeringAngleDeg = cp.vl["STEERING"]["STEERING_ANGLE"]
event.pressed = pressed
events.append(event)
self.button_states[button.event_type] = pressed
self.button_events = events

6
iqdbc_repo/iqdbc/iqpilot/car/chrysler/values_ext.py
View File

@@ -1,7 +1,9 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""
IQ.Pilot Chrysler extension flags + the cruise-button table read by the AOL
cruise-button reader.
"""
from collections import namedtuple
from enum import IntFlag
@@ -20,5 +22,3 @@ BUTTONS = [
class ChryslerFlagsIQ(IntFlag):
NO_MIN_STEERING_SPEED = 1

3
iqdbc_repo/iqdbc/iqpilot/car/ford/__init__.py
View File

@@ -1,3 +0,0 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""

21
iqdbc_repo/iqdbc/iqpilot/car/ford/aol.py
View File

@@ -1,21 +0,0 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""
from enum import StrEnum
from iqdbc.car import Bus,structs
from iqdbc.iqpilot.aol_base import AolCarStateBase
from iqdbc.can.parser import CANParser
class AolCarState(AolCarStateBase):
def __init__(self, CP: structs.CarParams, CP_IQ: structs.IQCarParams):
super().__init__(CP, CP_IQ)
def update_aol(self, ret: structs.CarState, can_parsers: dict[StrEnum, CANParser]) -> None:
cp = can_parsers[Bus.pt]
self.prev_lkas_button = self.lkas_button
self.lkas_button = cp.vl["Steering_Data_FD1"]["TjaButtnOnOffPress"]

3
iqdbc_repo/iqdbc/iqpilot/car/gm/__init__.py
View File

@@ -1,3 +0,0 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""

13
iqdbc_repo/iqdbc/iqpilot/car/gm/carstate_ext.py
View File

@@ -1,5 +1,8 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
Non-ACC GM carState: these cars have no adaptive cruise, so cruise engage/set
speed come from the stock (non-adaptive) ECM cruise message.
"""
from enum import StrEnum
@@ -15,9 +18,9 @@ class CarStateExt:
self.CP_IQ = CP_IQ
def update(self, ret: structs.CarState, can_parsers: dict[StrEnum, CANParser]) -> None:
if not self.CP_IQ.flags & GMFlagsIQ.NON_ACC:
return
pt_cp = can_parsers[Bus.pt]
if self.CP_IQ.flags & GMFlagsIQ.NON_ACC:
ret.cruiseState.enabled = pt_cp.vl["ECMCruiseControl"]["CruiseActive"] != 0
ret.cruiseState.speed = pt_cp.vl["ECMCruiseControl"]["CruiseSetSpeed"] * CV.KPH_TO_MS
ret.accFaulted = False
ret.cruiseState.enabled = pt_cp.vl["ECMCruiseControl"]["CruiseActive"] != 0
ret.cruiseState.speed = pt_cp.vl["ECMCruiseControl"]["CruiseSetSpeed"] * CV.KPH_TO_MS
ret.accFaulted = False

57
iqdbc_repo/iqdbc/iqpilot/car/gm/interface_ext.py
View File

@@ -1,18 +1,19 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
GM torque-space feed-forward for the IQ.Pilot lateral extension: a sigmoid+linear
lat-accel -> torque curve for the tuned platforms, else the linear default.
"""
import os
from math import exp
from iqdbc.car import structs
from iqdbc.car.common.basedir import BASEDIR
from iqdbc.car.gm.interface import CAR
from iqdbc.iqpilot.car.interfaces import LatControlInputs, NanoFFModel, TorqueFromLateralAccelCallbackTypeTorqueSpace
from iqdbc.iqpilot.car.interfaces import LatControlInputs, TorqueFromLateralAccelCallbackTypeTorqueSpace
NON_LINEAR_TORQUE_PARAMS = {
_NON_LINEAR_TORQUE_PARAMS = {
CAR.CHEVROLET_BOLT_EUV: [2.6531724862969748, 1.0, 0.1919764879840985, 0.009054123646805178],
CAR.GMC_ACADIA: [4.78003305, 1.0, 0.3122, 0.05591772],
CAR.CHEVROLET_SILVERADO: [3.29974374, 1.0, 0.25571356, 0.0465122]
CAR.CHEVROLET_SILVERADO: [3.29974374, 1.0, 0.25571356, 0.0465122],
}
@@ -20,39 +21,23 @@ class CarInterfaceExt:
def __init__(self, CP: structs.CarParams, CI_Base):
self.CP = CP
self.CI_Base = CI_Base
self.neural_ff_model = None
def torque_from_lateral_accel_siglin(self, latcontrol_inputs: LatControlInputs, torque_params: structs.CarParams.LateralTorqueTuning,
@staticmethod
def _centered_sigmoid(val: float) -> float:
# sigmoid shifted to pass through the origin; branch keeps exp() from overflowing
if val >= 0:
return 1.0 / (1.0 + exp(-val)) - 0.5
z = exp(val)
return z / (1.0 + z) - 0.5
def torque_from_lateral_accel_siglin(self, latcontrol_inputs: LatControlInputs,
torque_params: structs.CarParams.LateralTorqueTuning,
gravity_adjusted: bool) -> float:
def sig(val):
# https://timvieira.github.io/blog/post/2014/02/11/exp-normalize-trick
if val >= 0:
return 1 / (1 + exp(-val)) - 0.5
else:
z = exp(val)
return z / (1 + z) - 0.5
# The "lat_accel vs torque" relationship is assumed to be the sum of "sigmoid + linear" curves
# An important thing to consider is that the slope at 0 should be > 0 (ideally >1)
# This has big effect on the stability about 0 (noise when going straight)
# ToDo: To generalize to other GMs, explore tanh function as the nonlinear
non_linear_torque_params = NON_LINEAR_TORQUE_PARAMS.get(self.CP.carFingerprint)
assert non_linear_torque_params, "The params are not defined"
a, b, c, _ = non_linear_torque_params
steer_torque = (sig(latcontrol_inputs.lateral_acceleration * a) * b) + (latcontrol_inputs.lateral_acceleration * c)
return float(steer_torque)
def torque_from_lateral_accel_neural(self, latcontrol_inputs: LatControlInputs, orque_params: structs.CarParams.LateralTorqueTuning,
gravity_adjusted: bool) -> float:
inputs = list(latcontrol_inputs)
if gravity_adjusted:
inputs[0] += inputs[1]
return float(self.neural_ff_model.predict(inputs))
a, b, c, _ = _NON_LINEAR_TORQUE_PARAMS[self.CP.carFingerprint]
lat_accel = latcontrol_inputs.lateral_acceleration
return float(self._centered_sigmoid(lat_accel * a) * b + lat_accel * c)
def torque_from_lateral_accel_in_torque_space(self) -> TorqueFromLateralAccelCallbackTypeTorqueSpace:
if self.CP.carFingerprint == CAR.CHEVROLET_BOLT_EUV:
return self.torque_from_lateral_accel_neural
elif self.CP.carFingerprint in NON_LINEAR_TORQUE_PARAMS:
if self.CP.carFingerprint in _NON_LINEAR_TORQUE_PARAMS:
return self.torque_from_lateral_accel_siglin
else:
return self.CI_Base.torque_from_lateral_accel_linear_in_torque_space
return self.CI_Base.torque_from_lateral_accel_linear_in_torque_space

5
iqdbc_repo/iqdbc/iqpilot/car/gm/values_ext.py
View File

@@ -1,7 +1,8 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""
IQ.Pilot GM extension flags for the non-adaptive-cruise (Non-ACC) camera-harness port.
"""
from enum import IntFlag
@@ -11,5 +12,3 @@ class GMFlagsIQ(IntFlag):
class GMSafetyFlagsIQ:
NON_ACC = 1

19
iqdbc_repo/iqdbc/iqpilot/car/interfaces.py
View File

@@ -96,16 +96,17 @@ def _initialize_coop_steering(CP: structs.CarParams, CP_IQ: structs.IQCarParams,
def _initialize_stop_and_go(CP: structs.CarParams, CP_IQ: structs.IQCarParams, params_dict: dict[str, str]) -> None:
if CP.brand == 'subaru' and not CP.flags & (SubaruFlags.GLOBAL_GEN2 | SubaruFlags.HYBRID):
stop_and_go = int(params_dict.get("SubaruStopAndGo", 0)) == 1
stop_and_go_manual_parking_brake = int(params_dict.get("SubaruStopAndGoManualParkingBrake", 0)) == 1
# Subaru stop-and-go; unsupported on gen2-global and hybrid platforms.
if CP.brand != 'subaru' or CP.flags & (SubaruFlags.GLOBAL_GEN2 | SubaruFlags.HYBRID):
return
if stop_and_go:
CP_IQ.flags |= SubaruFlagsIQ.STOP_AND_GO.value
if stop_and_go_manual_parking_brake:
CP_IQ.flags |= SubaruFlagsIQ.STOP_AND_GO_MANUAL_PARKING_BRAKE.value
if stop_and_go or stop_and_go_manual_parking_brake:
CP_IQ.safetyParam |= SubaruSafetyFlagsIQ.STOP_AND_GO
if int(params_dict.get("SubaruStopAndGo", 0)) == 1:
CP_IQ.flags |= SubaruFlagsIQ.STOP_AND_GO.value
if int(params_dict.get("SubaruStopAndGoManualParkingBrake", 0)) == 1:
CP_IQ.flags |= SubaruFlagsIQ.STOP_AND_GO_MANUAL_PARKING_BRAKE.value
if CP_IQ.flags & (SubaruFlagsIQ.STOP_AND_GO | SubaruFlagsIQ.STOP_AND_GO_MANUAL_PARKING_BRAKE):
CP_IQ.iqSafetyFlags |= SubaruSafetyFlagsIQ.STOP_AND_GO
def _initialize_toyota(CP: structs.CarParams, CP_IQ: structs.IQCarParams, params_dict: dict[str, str]) -> None:

3
iqdbc_repo/iqdbc/iqpilot/car/mazda/__init__.py
View File

@@ -1,3 +0,0 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""

3
iqdbc_repo/iqdbc/iqpilot/car/nissan/__init__.py
View File

@@ -1,3 +0,0 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""

23
iqdbc_repo/iqdbc/iqpilot/car/nissan/carstate_ext.py
View File

@@ -1,7 +1,8 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""
Nissan cruise-button reader: emits edge ButtonEvents for the RES/SET buttons.
"""
from enum import StrEnum
from iqdbc.car import Bus, structs
@@ -13,20 +14,18 @@ class CarStateExt:
def __init__(self, CP, CP_IQ):
self.CP = CP
self.CP_IQ = CP_IQ
self.button_events = []
self.button_events: list = []
self.button_states = {button.event_type: False for button in BUTTONS}
def update(self, ret: structs.CarState, ret_iq: structs.IQCarState, can_parsers: dict[StrEnum, CANParser]):
def update(self, ret: structs.CarState, ret_iq: structs.IQCarState, can_parsers: dict[StrEnum, CANParser]) -> None:
cp = can_parsers[Bus.pt]
button_events = []
events = []
for button in BUTTONS:
state = (cp.vl[button.can_addr][button.can_msg] in button.values)
if self.button_states[button.event_type] != state:
pressed = cp.vl[button.can_addr][button.can_msg] in button.values
if pressed != self.button_states[button.event_type]:
event = structs.CarState.ButtonEvent.new_message()
event.type = button.event_type
event.pressed = state
button_events.append(event)
self.button_states[button.event_type] = state
self.button_events = button_events
event.pressed = pressed
events.append(event)
self.button_states[button.event_type] = pressed
self.button_events = events

7
iqdbc_repo/iqdbc/iqpilot/car/nissan/values.py
View File

@@ -1,8 +1,10 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""
IQ.Pilot Nissan extension: safety flag + cruise-button table.
"""
from collections import namedtuple
from iqdbc.car import structs
@@ -11,12 +13,9 @@ class NissanSafetyFlagsIQ:
LEAF = 1
ButtonType = structs.CarState.ButtonEvent.Type
Button = namedtuple('Button', ['event_type', 'can_addr', 'can_msg', 'values'])
BUTTONS = [
Button(ButtonType.accelCruise, "CRUISE_THROTTLE", "RES_BUTTON", [1]),
Button(ButtonType.decelCruise, "CRUISE_THROTTLE", "SET_BUTTON", [1]),

3
iqdbc_repo/iqdbc/iqpilot/car/rivian/__init__.py
View File

@@ -1,3 +0,0 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""

31
iqdbc_repo/iqdbc/iqpilot/car/rivian/aol.py
View File

@@ -1,33 +1,30 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
Always-on-Lateral output adapter for Rivian: derives the per-frame lateral-active
and lane-keep icon state the LKAS command needs from the shared AOL state.
"""
from collections import namedtuple
from iqdbc.car import structs
from iqdbc.car.interfaces import CarStateBase
MAX_STEERING_ANGLE = 90.0
# Rivian EPAS rejects large-angle torque, so lateral is dropped past this bound.
_MAX_STEERING_ANGLE = 90.0
AolDataIQ = namedtuple("AolDataIQ",
["lka_icon_states", "lat_active"])
AolDataIQ = namedtuple("AolDataIQ", ["lka_icon_states", "lat_active"])
class AolCarController:
def __init__(self):
self.aol = AolDataIQ(False, False)
self.lka_icon_states = False
self.lat_active = False
def aol_status_update(self, CC: structs.CarControl, CC_IQ: structs.IQCarControl, CS: CarStateBase) -> AolDataIQ:
if CC_IQ.aol.available:
self.lka_icon_states = self.lat_active
self.lat_active = CC.latActive and abs(CS.out.steeringAngleDeg) < MAX_STEERING_ANGLE
else:
self.lka_icon_states = CC.enabled
self.lat_active = CC.latActive
return AolDataIQ(self.lka_icon_states, self.lat_active)
def update(self, CC: structs.CarControl, CC_IQ: structs.IQCarControl, CS: CarStateBase) -> None:
self.aol = self.aol_status_update(CC, CC_IQ, CS)
prev_active = self.aol.lat_active
if CC_IQ.aol.available:
lat_active = CC.latActive and abs(CS.out.steeringAngleDeg) < _MAX_STEERING_ANGLE
lka_icon = prev_active
else:
lat_active = CC.latActive
lka_icon = CC.enabled
self.aol = AolDataIQ(lka_icon, lat_active)

76
iqdbc_repo/iqdbc/iqpilot/car/rivian/carstate_ext.py
View File

@@ -1,5 +1,9 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
Rivian longitudinal-harness-upgrade carState reader: with the upgrade harness the
right steering-wheel controls and the drive stalk drive the openpilot set speed,
and the harness exposes blind-spot indicators. Only active behind the upgrade flag.
"""
import math
from enum import StrEnum
@@ -12,8 +16,10 @@ from iqdbc.iqpilot.car.rivian.values import RivianFlagsIQ
ButtonType = structs.CarState.ButtonEvent.Type
MAX_SET_SPEED = 85 * CV.MPH_TO_MS
MIN_SET_SPEED = 20 * CV.MPH_TO_MS
_SET_SPEED_MAX = 85 * CV.MPH_TO_MS
_SET_SPEED_MIN = 20 * CV.MPH_TO_MS
_LONG_PRESS_FRAMES = 66
_STALK_HOLD_FRAMES = 50
class CarStateExt:
@@ -29,57 +35,57 @@ class CarStateExt:
self.decrease_counter = 0
self.stalk_down_counter = 0
def _apply_set_speed_buttons(self, ret: structs.CarState, cp_park, cp_adas) -> None:
was_increasing = self.increase_button
was_decreasing = self.decrease_button
self.increase_button = cp_park.vl["WheelButtons"]["RightButton_RightClick"] == 2
self.decrease_button = cp_park.vl["WheelButtons"]["RightButton_LeftClick"] == 2
self.increase_counter = self.increase_counter + 1 if self.increase_button else 0
self.decrease_counter = self.decrease_counter + 1 if self.decrease_button else 0
metric = cp_adas.vl["Cluster"]["Cluster_Unit"] == 0
conversion = CV.KPH_TO_MS if metric else CV.MPH_TO_MS
step = 10.0 if metric else 5.0
shown = self.set_speed * (CV.MS_TO_KPH if metric else CV.MS_TO_MPH)
# A held button steps to the next round multiple; a tap nudges by one unit.
if self.increase_button:
if self.increase_counter % _LONG_PRESS_FRAMES == 0:
self.set_speed = math.ceil((shown + 1) / step) * step * conversion
elif not was_increasing:
self.set_speed += conversion
if self.decrease_button:
if self.decrease_counter % _LONG_PRESS_FRAMES == 0:
self.set_speed = math.floor((shown - 1) / step) * step * conversion
elif not was_decreasing:
self.set_speed -= conversion
def update_longitudinal_upgrade(self, ret: structs.CarState, can_parsers: dict[StrEnum, CANParser]) -> None:
cp_park = can_parsers[Bus.alt]
cp_adas = can_parsers[Bus.adas]
cp = can_parsers[Bus.pt]
prev_increase_button = self.increase_button
prev_decrease_button = self.decrease_button
if self.CP.openpilotLongitudinalControl:
# distance scroll wheel
right_scroll = cp_park.vl["WheelButtons"]["RightButton_Scroll"]
if right_scroll != 255:
if self.distance_button != right_scroll:
ret.buttonEvents = [structs.CarState.ButtonEvent(pressed=False, type=ButtonType.gapAdjustCruise)]
self.distance_button = right_scroll
# button logic for set-speed
self.increase_button = cp_park.vl["WheelButtons"]["RightButton_RightClick"] == 2
self.decrease_button = cp_park.vl["WheelButtons"]["RightButton_LeftClick"] == 2
self.increase_counter = self.increase_counter + 1 if self.increase_button else 0
self.decrease_counter = self.decrease_counter + 1 if self.decrease_button else 0
metric = cp_adas.vl["Cluster"]["Cluster_Unit"] == 0
conversion = CV.KPH_TO_MS if metric else CV.MPH_TO_MS
long_press_step = 10.0 if metric else 5.0
set_speed_converted = self.set_speed * (CV.MS_TO_KPH if metric else CV.MS_TO_MPH)
if self.increase_button:
if self.increase_counter % 66 == 0:
self.set_speed = (int(math.ceil((set_speed_converted + 1) / long_press_step)) * long_press_step) * conversion
elif not prev_increase_button:
self.set_speed += conversion
if self.decrease_button:
if self.decrease_counter % 66 == 0:
self.set_speed = (int(math.floor((set_speed_converted - 1) / long_press_step)) * long_press_step) * conversion
elif not prev_decrease_button:
self.set_speed -= conversion
self._apply_set_speed_buttons(ret, cp_park, cp_adas)
if not ret.cruiseState.enabled:
self.set_speed = ret.vEgoCluster
# VDM_UserAdasRequest: 0=IDLE, 1=UP_1, 2=UP_2, 3=DOWN_1, 4=DOWN_2
# Drive stalk held down (VDM_UserAdasRequest 3/4) for ~0.5s snaps set speed
# up to the current speed, matching stock Rivian ACC (it never lowers it).
stalk_down = int(cp.vl["VDM_AdasSts"]["VDM_UserAdasRequest"]) in (3, 4)
self.stalk_down_counter = self.stalk_down_counter + 1 if stalk_down else 0
if self.stalk_down_counter == 50:
# Mimic Rivian ACC: holding stalk 0.5s sets speed to current speed (never decreases)
if self.stalk_down_counter == _STALK_HOLD_FRAMES:
self.set_speed = max(self.set_speed, ret.vEgoCluster)
self.set_speed = max(MIN_SET_SPEED, min(self.set_speed, MAX_SET_SPEED))
self.set_speed = max(_SET_SPEED_MIN, min(self.set_speed, _SET_SPEED_MAX))
ret.cruiseState.speed = self.set_speed
if self.CP.enableBsm:
@@ -92,9 +98,7 @@ class CarStateExt:
@staticmethod
def get_parser(CP, CP_IQ) -> dict[StrEnum, CANParser]:
messages = {}
messages: dict[StrEnum, CANParser] = {}
if CP_IQ.flags & RivianFlagsIQ.LONGITUDINAL_HARNESS_UPGRADE:
messages[Bus.alt] = CANParser(DBC[CP.carFingerprint][Bus.alt], [], 5)
return messages

4
iqdbc_repo/iqdbc/iqpilot/car/rivian/values.py
View File

@@ -1,10 +1,10 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
IQ.Pilot Rivian extension flags.
"""
from enum import IntFlag
class RivianFlagsIQ(IntFlag):
LONGITUDINAL_HARNESS_UPGRADE = 1

3
iqdbc_repo/iqdbc/iqpilot/car/subaru/__init__.py
View File

@@ -1,3 +0,0 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""

49
iqdbc_repo/iqdbc/iqpilot/car/subaru/aol.py
View File

@@ -1,45 +1,36 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
Always-on-Lateral input adapter for Subaru: turns the dashboard LKAS button into
a carState lkas ButtonEvent so the shared AOL state machine can toggle lateral.
"""
from enum import StrEnum
from iqdbc.car import Bus, structs
from iqdbc.car import Bus, structs
from iqdbc.car.subaru.values import SubaruFlags
from iqdbc.iqpilot.aol_base import AolCarStateBase
from iqdbc.can.parser import CANParser
ButtonType = structs.CarState.ButtonEvent.Type
_LKAS = structs.CarState.ButtonEvent.Type.lkas
# ES_LKAS_State/LKAS_Dash_State: 0 = neutral, 1 = LKAS shown on, 2 = LKAS shown off
_DASH_ON = 1
_DASH_OFF = 2
class AolCarState(AolCarStateBase):
def __init__(self, CP: structs.CarParams, CP_IQ: structs.IQCarParams):
super().__init__(CP, CP_IQ)
@staticmethod
def create_lkas_button_events(cur_btn: int, prev_btn: int,
buttons_dict: dict[int, structs.CarState.ButtonEvent.Type]) -> list[structs.CarState.ButtonEvent]:
events: list[structs.CarState.ButtonEvent] = []
if cur_btn == prev_btn:
return events
state_changes = [
{"pressed": prev_btn != cur_btn and cur_btn != 2 and not (prev_btn == 2 and cur_btn == 1)},
{"pressed": prev_btn != cur_btn and cur_btn == 2 and cur_btn != 1},
]
for change in state_changes:
if change["pressed"]:
events.append(structs.CarState.ButtonEvent(pressed=change["pressed"],
type=buttons_dict.get(cur_btn, ButtonType.unknown)))
return events
def update_aol(self, ret: structs.CarState, can_parsers: dict[StrEnum, CANParser]) -> None:
cp_cam = can_parsers[Bus.cam]
if self.CP.flags & SubaruFlags.PREGLOBAL:
return
self.prev_lkas_button = self.lkas_button
if not self.CP.flags & SubaruFlags.PREGLOBAL:
self.lkas_button = cp_cam.vl["ES_LKAS_State"]["LKAS_Dash_State"]
self.lkas_button = can_parsers[Bus.cam].vl["ES_LKAS_State"]["LKAS_Dash_State"]
ret.buttonEvents = self.create_lkas_button_events(self.lkas_button, self.prev_lkas_button, {1: ButtonType.lkas})
if self._is_toggle_edge():
ret.buttonEvents = [*ret.buttonEvents, structs.CarState.ButtonEvent(type=_LKAS, pressed=True)]
def _is_toggle_edge(self) -> bool:
# every dash-state change is a deliberate press except the off->on rebound (2 -> 1)
if self.lkas_button == self.prev_lkas_button:
return False
return not (self.prev_lkas_button == _DASH_OFF and self.lkas_button == _DASH_ON)

108
iqdbc_repo/iqdbc/iqpilot/car/subaru/stop_and_go.py
View File

@@ -1,7 +1,11 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
"""
IQ.Pilot Subaru stop-and-go: nudges the ACC out of a standstill it would
otherwise hold. Two variants gated by user flags — an electronic-parking-brake
resume pulse (distance/lead triggered) and a manual-parking-brake hold-timer
resume. Both work by spoofing the camera-bus throttle/brake frames.
"""
import copy
from enum import StrEnum
@@ -14,93 +18,64 @@ from iqdbc.iqpilot.car.subaru import subarucan_ext
from iqdbc.iqpilot.car.subaru.values_ext import SubaruFlagsIQ
from iqdbc.can.parser import CANParser
_SNG_ACC_MIN_DIST = 3
_SNG_ACC_MAX_DIST = 4.5
# EPB resume fires only while the lead is pulling away within this gap band (m).
_RESUME_GAP_MIN = 3.0
_RESUME_GAP_MAX = 4.5
_EPB_PULSE_FRAMES = 15
class SnGCarController:
def __init__(self, CP: structs.CarParams, CP_IQ: structs.IQCarParams):
self.CP = CP
self.CP_IQ = CP_IQ
self.enabled = CP_IQ.flags & (SubaruFlagsIQ.STOP_AND_GO | SubaruFlagsIQ.STOP_AND_GO_MANUAL_PARKING_BRAKE)
self.manual_parking_brake = CP_IQ.flags & SubaruFlagsIQ.STOP_AND_GO_MANUAL_PARKING_BRAKE
self.enabled = bool(CP_IQ.flags & (SubaruFlagsIQ.STOP_AND_GO | SubaruFlagsIQ.STOP_AND_GO_MANUAL_PARKING_BRAKE))
self.manual_parking_brake = bool(CP_IQ.flags & SubaruFlagsIQ.STOP_AND_GO_MANUAL_PARKING_BRAKE)
self.last_standstill_frame = 0
self.epb_resume_frames_remaining = -1
self.prev_close_distance = 0.0
self.standstill_since = 0
self._pulse_left = 0
self.prev_gap = 0.0
def update_epb_resume_sequence(self, should_resume: bool) -> bool:
def _epb_pulse(self, trigger: bool) -> bool:
# A trigger arms a fixed-length resume pulse; the pulse then plays out frame by frame.
if self.manual_parking_brake:
return False
if trigger:
self._pulse_left = _EPB_PULSE_FRAMES
if self._pulse_left > 0:
self._pulse_left -= 1
return True
return False
if should_resume:
self.epb_resume_frames_remaining = 15
send_resume = self.epb_resume_frames_remaining > 0
if self.epb_resume_frames_remaining > 0:
self.epb_resume_frames_remaining -= 1
return send_resume
def update_stop_and_go(self, CC: structs.CarControl, CS: CarStateBase, frame: int) -> bool:
"""
Manages stop-and-go functionality for adaptive cruise control (ACC).
Args:
CC: Car control data
CS: Car state data
frame: Current frame number
Returns:
bool: True if resume command should be sent, False otherwise
"""
def _want_resume(self, CC: structs.CarControl, CS: CarStateBase, frame: int) -> bool:
if not CC.enabled or not CC.hudControl.leadVisible:
return False
close_distance = CS.es_distance_msg["Close_Distance"]
in_standstill = CS.out.standstill
gap = CS.es_distance_msg["Close_Distance"]
standing = CS.out.standstill
if not standing:
self.standstill_since = frame
if not in_standstill:
self.last_standstill_frame = frame
hold_arm, hold_reset = (0.75, 0.8) if self.CP.flags & SubaruFlags.PREGLOBAL else (0.5, 0.55)
held_for = (frame - self.standstill_since) * DT_CTRL
held_long_enough = held_for > hold_arm
if held_for >= hold_reset:
self.standstill_since = frame
# Check if we've been in standstill long enough
mpb_standstill_timers = (0.75, 0.8) if self.CP.flags & SubaruFlags.PREGLOBAL else (0.5, 0.55)
standstill_duration = (frame - self.last_standstill_frame) * DT_CTRL
in_standstill_hold = standstill_duration > mpb_standstill_timers[0]
if (frame - self.last_standstill_frame) * DT_CTRL >= mpb_standstill_timers[1]:
self.last_standstill_frame = frame
# Car state distance-based conditions (EPB only)
in_resume_distance = _SNG_ACC_MIN_DIST < close_distance < _SNG_ACC_MAX_DIST
distance_increasing = close_distance > self.prev_close_distance
distance_resume_allowed = in_resume_distance and distance_increasing
lead_pulling_away = _RESUME_GAP_MIN < gap < _RESUME_GAP_MAX and gap > self.prev_gap
self.prev_gap = gap
if self.manual_parking_brake:
# Manual parking brake: Direct resume when the standstill hold threshold is reached to prevent ACC fault
send_resume = in_standstill_hold
else:
# EPB: Resume sequence with trigger on distance with lead car increasing
should_resume = CS.out.standstill and distance_resume_allowed
send_resume = self.update_epb_resume_sequence(should_resume)
self.prev_close_distance = close_distance
return send_resume
return held_long_enough
return self._epb_pulse(standing and lead_pulling_away)
def create_stop_and_go(self, packer, CC: structs.CarControl, CS: CarStateBase, frame: int) -> list[CanData]:
can_sends = []
if not self.enabled:
return can_sends
send_resume = self.update_stop_and_go(CC, CS, frame)
can_sends.append(subarucan_ext.create_throttle(packer, self.CP, CS.throttle_msg, send_resume and not self.manual_parking_brake))
return []
resume = self._want_resume(CC, CS, frame)
can_sends = [subarucan_ext.create_throttle(packer, self.CP, CS.throttle_msg, resume and not self.manual_parking_brake)]
if frame % 2 == 0:
can_sends.append(subarucan_ext.create_brake_pedal(packer, self.CP, CS.brake_pedal_msg, send_resume and self.manual_parking_brake))
can_sends.append(subarucan_ext.create_brake_pedal(packer, self.CP, CS.brake_pedal_msg, resume and self.manual_parking_brake))
return can_sends
@@ -108,14 +83,11 @@ class SnGCarState:
def __init__(self, CP: structs.CarParams, CP_IQ: structs.IQCarParams):
self.CP = CP
self.CP_IQ = CP_IQ
self.brake_pedal_msg: dict[str, float] = {}
self.throttle_msg: dict[str, float] = {}
def update(self, ret: structs.CarState, can_parsers: dict[StrEnum, CANParser]) -> None:
cp = can_parsers[Bus.pt]
self.brake_pedal_msg = copy.copy(cp.vl["Brake_Pedal"])
if not self.CP.flags & SubaruFlags.HYBRID:
self.throttle_msg = copy.copy(cp.vl["Throttle"])

79
iqdbc_repo/iqdbc/iqpilot/car/subaru/subarucan_ext.py
View File

@@ -1,68 +1,41 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
Camera-bus message spoofers for Subaru stop-and-go: re-emit the stock Throttle
and Brake_Pedal frames, nudging a single field to trigger an ACC resume from a
standstill. Field sets mirror the DBC message layout.
"""
from iqdbc.car.subaru.values import CanBus, SubaruFlags
_THROTTLE_FIELDS_PREGLOBAL = ("Throttle_Pedal", "Signal1", "Not_Full_Throttle", "Signal2", "Engine_RPM",
"Off_Throttle", "Signal3", "Throttle_Cruise", "Throttle_Combo", "Throttle_Body",
"Off_Throttle_2", "Signal4")
_THROTTLE_FIELDS_GLOBAL = ("CHECKSUM", "Signal1", "Engine_RPM", "Neutral", "Throttle_Pedal", "Throttle_Cruise",
"Throttle_Combo", "Signal3", "Off_Accel")
_BRAKE_FIELDS_PREGLOBAL = ("Speed", "Brake_Pedal", "Signal1")
_BRAKE_FIELDS_GLOBAL = ("CHECKSUM", "Signal1", "Speed", "Signal2", "Brake_Lights", "Signal3", "Brake_Pedal", "Signal4")
def create_counter(msg):
def _next_counter(msg):
return (msg["COUNTER"] + 1) % 0x10
def create_throttle(packer, CP, throttle_msg, send_resume):
if CP.flags & SubaruFlags.PREGLOBAL:
values = {s: throttle_msg[s] for s in [
"Throttle_Pedal",
"Signal1",
"Not_Full_Throttle",
"Signal2",
"Engine_RPM",
"Off_Throttle",
"Signal3",
"Throttle_Cruise",
"Throttle_Combo",
"Throttle_Body",
"Off_Throttle_2",
"Signal4",
]}
else:
values = {s: throttle_msg[s] for s in [
"CHECKSUM",
"Signal1",
"Engine_RPM",
"Neutral",
"Throttle_Pedal",
"Throttle_Cruise",
"Throttle_Combo",
"Signal3",
"Off_Accel",
]}
values["COUNTER"] = create_counter(throttle_msg)
preglobal = bool(CP.flags & SubaruFlags.PREGLOBAL)
fields = _THROTTLE_FIELDS_PREGLOBAL if preglobal else _THROTTLE_FIELDS_GLOBAL
values = {name: throttle_msg[name] for name in fields}
values["COUNTER"] = _next_counter(throttle_msg)
if send_resume:
values["Throttle_Pedal"] = 5
return packer.make_can_msg("Throttle", CanBus.camera, values)
def create_brake_pedal(packer, CP, brake_pedal_msg, send_resume):
if CP.flags & SubaruFlags.PREGLOBAL:
values = {s: brake_pedal_msg[s] for s in [
"Speed",
"Brake_Pedal",
"Signal1",
]}
else:
values = {s: brake_pedal_msg[s] for s in [
"CHECKSUM",
"Signal1",
"Speed",
"Signal2",
"Brake_Lights",
"Signal3",
"Brake_Pedal",
"Signal4",
]}
values["COUNTER"] = create_counter(brake_pedal_msg)
preglobal = bool(CP.flags & SubaruFlags.PREGLOBAL)
fields = _BRAKE_FIELDS_PREGLOBAL if preglobal else _BRAKE_FIELDS_GLOBAL
values = {name: brake_pedal_msg[name] for name in fields}
if not preglobal:
values["COUNTER"] = _next_counter(brake_pedal_msg)
if send_resume:
values["Speed"] = 1 if CP.flags & SubaruFlags.PREGLOBAL else 3
values["Speed"] = 1 if preglobal else 3
return packer.make_can_msg("Brake_Pedal", CanBus.camera, values)

12
iqdbc_repo/iqdbc/iqpilot/car/subaru/values_ext.py
View File

@@ -1,14 +1,16 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
IQ.Pilot Subaru extension flags. Selected in setup_interfaces() from the user's
stop-and-go params and consumed by the stop-and-go controller and panda safety.
"""
from enum import IntFlag
class SubaruSafetyFlagsIQ:
STOP_AND_GO = 1
class SubaruFlagsIQ(IntFlag):
STOP_AND_GO = 1
STOP_AND_GO_MANUAL_PARKING_BRAKE = 2
class SubaruSafetyFlagsIQ:
STOP_AND_GO = 1

5
iqdbc_repo/iqdbc/safety/tests/test_gm.py
View File

@@ -7,7 +7,8 @@ from iqdbc.safety.tests.libsafety import libsafety_py
import iqdbc.safety.tests.common as common
from iqdbc.safety.tests.common import CANPackerSafety
from iqdbc.iqpilot.car.gm.values_ext import GMSafetyFlagsIQ
# GM_PARAM_IQ_NON_ACC in safety/modes/gm.h (IQ safety framework flag)
GM_PARAM_IQ_NON_ACC = 1
class Buttons:
@@ -236,7 +237,7 @@ class TestGmCameraNonACCSafety(TestGmCameraSafety):
self.packer = CANPackerSafety("gm_global_a_powertrain_generated")
self.packer_chassis = CANPackerSafety("gm_global_a_chassis")
self.safety = libsafety_py.libsafety
self.safety.set_current_safety_param_iq(GMSafetyFlagsIQ.NON_ACC)
self.safety.set_current_safety_param_iq(GM_PARAM_IQ_NON_ACC)
self.safety.set_safety_hooks(CarParams.SafetyModel.gm, GMSafetyFlags.HW_CAM | self.EXTRA_SAFETY_PARAM)
self.safety.init_tests()

6
iqdbc_repo/iqdbc/safety/tests/test_nissan.py
View File

@@ -2,8 +2,10 @@
import unittest
from iqdbc.car.nissan.values import NissanSafetyFlags
from iqdbc.iqpilot.car.nissan.values import NissanSafetyFlagsIQ
from iqdbc.car.structs import CarParams
# NISSAN_PARAM_IQ_LEAF in safety/modes/nissan.h (IQ safety framework flag)
NISSAN_PARAM_IQ_LEAF = 1
from iqdbc.safety.tests.libsafety import libsafety_py
import iqdbc.safety.tests.common as common
from iqdbc.safety.tests.common import CANPackerSafety
@@ -105,7 +107,7 @@ class TestNissanLeafSafety(TestNissanSafety):
def setUp(self):
self.packer = CANPackerSafety("nissan_leaf_2018_generated")
self.safety = libsafety_py.libsafety
self.safety.set_current_safety_param_iq(NissanSafetyFlagsIQ.LEAF)
self.safety.set_current_safety_param_iq(NISSAN_PARAM_IQ_LEAF)
self.safety.set_safety_hooks(CarParams.SafetyModel.nissan, 0)
self.safety.init_tests()