Merge gateway MEB support and qualified Toyota replay metadata

This commit is contained in:
firestar5683
2026-10-22 22:48:54 -05:00
66 changed files with 1736 additions and 119 deletions
@@ -146,7 +146,8 @@ class CarController(CarControllerBase):
if self.CP.flags & VolkswagenFlags.MEB:
accel = float(np.clip(actuators.accel, self.CCP.ACCEL_MIN, self.CCP.ACCEL_MAX))
accel, acc_status, acc_hold_type, braking_to_stop, leaving_standstill = self.meb_long_state.update(CS, CC, accel)
can_sends.extend(mebcan.create_acc_accel_control(self.packer_pt, self.CAN.pt, self.CCP, CS.acc_type, CC.enabled,
acc_enabled = self.meb_long_state._long_active(CS, CC)
can_sends.extend(mebcan.create_acc_accel_control(self.packer_pt, self.CAN.pt, self.CCP, CS.acc_type, acc_enabled,
accel, acc_status, acc_hold_type, braking_to_stop, leaving_standstill,
CS.out.vEgoRaw * CV.MS_TO_KPH, CS.travel_assist_available))
@@ -204,7 +205,8 @@ class CarController(CarControllerBase):
gra_send_ready = self.CP.pcmCruise and CS.gra_stock_values["COUNTER"] != self.gra_acc_counter_last
if gra_send_ready and (CC.cruiseControl.cancel or CC.cruiseControl.resume):
can_sends.append(self.CCS.create_acc_buttons_control(self.packer_pt, self.CAN.ext, CS.gra_stock_values,
button_control = mebcan.create_acc_buttons_control if self.CP.flags & VolkswagenFlags.MEB else self.CCS.create_acc_buttons_control
can_sends.append(button_control(self.packer_pt, self.CAN.ext, CS.gra_stock_values,
cancel=CC.cruiseControl.cancel, resume=CC.cruiseControl.resume))
new_actuators = actuators.as_builder()
@@ -299,6 +299,11 @@ class CarState(CarStateBase):
ret.cruiseState.available = pt_cp.vl["Motor_51"]["TSK_Status"] in (2, 3, 4, 5)
ret.cruiseState.enabled = pt_cp.vl["Motor_51"]["TSK_Status"] in (3, 4, 5)
ret.cruiseState.nonAdaptive = bool(pt_cp.vl["Motor_51"]["TSK_Limiter_ausgewaehlt"])
if self.CP.pcmCruise:
ret.cruiseState.speed = ext_cp.vl["ACC_19"]["ACC_Wunschgeschw_02"] * CV.KPH_TO_MS
if ret.cruiseState.speed > 90:
ret.cruiseState.speed = 0
ret.cruiseState.standstill = ret.standstill
tsk_faulted = pt_cp.vl["Motor_51"]["TSK_Status"] in (6, 7)
engine_off = pt_cp.vl["Motor_54"]["Engine_On"] == 0
@@ -124,6 +124,7 @@ class CarInterface(CarInterfaceBase):
if ret.flags & VolkswagenFlags.MEB:
ret.openpilotLongitudinalControl = True
ret.alphaLongitudinalAvailable = ret.networkLocation == NetworkLocation.gateway or docs
ret.longitudinalActuatorDelay = 0.3
else:
ret.alphaLongitudinalAvailable = ret.networkLocation == NetworkLocation.gateway or docs
@@ -0,0 +1,58 @@
"""Source-owned MEB stock ACC option; upstream system-long remains the default."""
from opendbc.car import structs
from opendbc.car.volkswagen.values import CAR, VolkswagenFlags
from opendbc.car.volkswagen.stock_aol import safety_config
STOCK_WORD = 4
AOL_WORD = 8
ALT_GEAR_WORD = 16
MEB_CARS = frozenset(c for c in CAR if c.config.flags & VolkswagenFlags.MEB)
KNOWN_FLAGS = int(VolkswagenFlags.MEB | VolkswagenFlags.MEB_GEN2 | VolkswagenFlags.ALT_GEAR |
VolkswagenFlags.STOCK_KLR_PRESENT | VolkswagenFlags.HAS_BSM)
AOL_WORDS = frozenset((8, 10, 12, 14, 24, 26, 28, 30))
def base_word(cp):
return (2 if cp.flags & VolkswagenFlags.MEB_GEN2 else 0) | (0 if cp.openpilotLongitudinalControl else STOCK_WORD)
def profile_word(cp):
return base_word(cp) | AOL_WORD | (ALT_GEAR_WORD if cp.flags & VolkswagenFlags.ALT_GEAR else 0)
def supported(cp):
try:
return bool(cp.brand == 'volkswagen' and cp.carFingerprint in MEB_CARS and
cp.flags & VolkswagenFlags.MEB and not cp.flags & ~KNOWN_FLAGS and not cp.notCar and not cp.dashcamOnly and
cp.networkLocation == structs.CarParams.NetworkLocation.gateway and
cp.transmissionType == structs.CarParams.TransmissionType.direct and
cp.steerControlType == structs.CarParams.SteerControlType.curvature and safety_config(cp) is not None and
safety_config(cp).safetyModel == structs.CarParams.SafetyModel.volkswagenMeb and
safety_config(cp).safetyParam in (base_word(cp), profile_word(cp)))
except (AttributeError, TypeError, ValueError, OverflowError, IndexError):
return False
def qualified(cp, *, marked_only=False):
try:
if not supported(cp) or cp.passive or cp.openpilotLongitudinalControl == cp.pcmCruise:
return False
word = int(safety_config(cp).safetyParam)
return ((cp.alternativeExperience == 32 and word == profile_word(cp)) or
(not marked_only and cp.alternativeExperience == 0 and word == base_word(cp)))
except (AttributeError, TypeError, ValueError, OverflowError, IndexError):
return False
def prepare_stock(cp, requested):
if requested and supported(cp):
cp.openpilotLongitudinalControl = False
cp.pcmCruise = True
safety_config(cp).safetyParam |= STOCK_WORD
def settings_capable(cp):
try:
return supported(cp) and not cp.passive
except AttributeError:
return False
+30 -3
View File
@@ -59,6 +59,24 @@ def create_lka_hud_control(packer, bus, ldw_stock_values, lat_active, steering_p
return packer.make_can_msg("LDW_02", bus, values)
def create_acc_buttons_control(packer, bus, gra_stock_values, cancel=False, resume=False, up=False, down=False):
values = {s: gra_stock_values[s] for s in [
"GRA_Hauptschalter", # ACC button, on/off
"GRA_Typ_Hauptschalter", # ACC main button type
"GRA_Codierung", # ACC button configuration/coding
"GRA_Tip_Stufe_2", # unknown related to stalk type
"GRA_ButtonTypeInfo", # unknown related to stalk type
]}
values.update({
"COUNTER": (gra_stock_values["COUNTER"] + 1) % 16,
"GRA_Abbrechen": cancel,
"GRA_Tip_Wiederaufnahme": resume or up,
"GRA_Tip_Setzen": down,
})
return packer.make_can_msg("GRA_ACC_01", bus, values)
ACC_HUD_ERROR = 6
ACC_HUD_OVERRIDE = 4
ACC_HUD_ACTIVE = 3
@@ -71,6 +89,7 @@ class MebLongStateMachine:
def __init__(self, CP, CCP):
self.CCP = CCP
self.CP = CP
self.RAMP_FRAMES = 10 // CCP.ACC_CONTROL_STEP # 100 ms
self.disengage_ramp_counter = 0 # always ramp when disengaging
@@ -82,12 +101,20 @@ class MebLongStateMachine:
self.prev_acc_hold_type = self.acc_hold_type_vals['KEINE_ANFORDERUNG'] # no request
self.acc_status = self.acc_status_vals['ACC_OFF_HAUPTSCHALTER_AUS'] # last acc status, read by HUD msg
def _independent_axes(self):
from opendbc.car.volkswagen.meb_modes import qualified
return qualified(self.CP, marked_only=True)
def _long_active(self, CS, CC):
return (CC.longActive and not CS.out.gasPressed and not CS.out.brakePressed
if self._independent_axes() else CC.enabled)
def _get_acc_status(self, CS, CC) -> int:
# stateless
# NOTE: stock TSK and camera goes to 5 on disengage independently which we don't model, but hasn't been shown to fault without it
if CS.out.accFaulted:
return self.acc_status_vals['REVERSIBLER_FEHLER_IM_ACC_SYSTEM']
elif CC.enabled:
elif self._long_active(CS, CC):
return self.acc_status_vals['ACC_OVERRIDE' if CC.cruiseControl.override else 'ACC_AKTIV_REGELT']
elif CS.out.cruiseState.available:
return self.acc_status_vals['ACC_STANDBY']
@@ -104,7 +131,7 @@ class MebLongStateMachine:
# HMS_Status leaves holding as soon as we send anfahren, so keep driving off until we leave standstill
driving_off = self.prev_acc_hold_type == self.acc_hold_type_vals['ANFAHREN'] and CS.esp_standstill
starting = CC.actuators.longControlState == LongCtrlState.pid and (CS.esp_hold_confirmation or driving_off)
long_active = CC.longActive and not CS.out.accFaulted # catches it one frame earlier, not sure if needed
long_active = (self._long_active(CS, CC) if self._independent_axes() else CC.longActive) and not CS.out.accFaulted
if not long_active:
# Stock goes to RAMP for as long as TSK_Status is 5 usually, 100ms seems fine to mimic that behavior.
@@ -144,7 +171,7 @@ class MebLongStateMachine:
# transition to inactive accel and jerks as soon as we enter ESP standstill
requesting_hold = acc_hold_type == self.acc_hold_type_vals['HALTEN']
held = requesting_hold and CS.esp_hold_confirmation
if not CC.enabled or held:
if not self._long_active(CS, CC) or held:
accel = self.CCP.ACCEL_INACTIVE
# hold requested but the car hasn't reached standstill yet
@@ -0,0 +1,66 @@
"""Public MEB get_car: transport fixtures, genuine FW matcher/factory/Card finalization."""
import os
from unittest import mock
import pytest
from opendbc.can import CANPacker
from opendbc.car import car_helpers, structs
from opendbc.car.can_definitions import CanData
from opendbc.car.fw_versions import match_fw_to_car
from opendbc.car.volkswagen.fingerprints import FW_VERSIONS
from opendbc.car.volkswagen.values import CAR, DBC
from opendbc.car import Bus
from openpilot.common.params import Params
from openpilot.common.prefix import OpenpilotPrefix
from openpilot.selfdrive.car.card import Car
@pytest.mark.parametrize('identity', (CAR.VOLKSWAGEN_ID4_MK1, CAR.VOLKSWAGEN_ID4_MK2))
@pytest.mark.parametrize('stock', (False, True))
@pytest.mark.parametrize('offset,fixed', ((0, False), (4, True)))
def test_normal_meb_discovery_and_final_card_owner(identity, stock, offset, fixed):
# Full synthetic ECU reply inventory from the current source DB; not an observed physical capture.
fw = [structs.CarParams.CarFw(ecu=ecu, address=address, subAddress=sub or 0, fwVersion=versions[0],
brand='volkswagen', bus=offset)
for (ecu, address, sub), versions in FW_VERSIONS[identity].items()]
vin = 'WVWZZZE2ZMP000000' if identity == CAR.VOLKSWAGEN_ID4_MK1 else 'WVWZZZE8ZRP000000'
exact, matches = match_fw_to_car(fw, vin, log=False)
assert exact == (identity == CAR.VOLKSWAGEN_ID4_MK2) and matches == {identity}
packer = CANPacker(DBC[identity][Bus.pt])
messages = [packer.make_can_msg('Getriebe_11', offset, {'GE_Fahrstufe': 8}),
packer.make_can_msg('QFK_01', offset + 1, {'LatCon_HCA_Status': 2})]
frames = [CanData(address=addr, dat=data, src=bus) for addr, data, bus in messages]
def receive(wait_for_one=False):
return [frames]
with OpenpilotPrefix(), mock.patch.dict(os.environ, {'SIMULATION': '0', 'AOL_REPLAY_RUNTIME': '1'}):
store = Params()
store.put_bool('AlwaysOnLateral', True, block=True)
store.put_bool('OpenpilotEnabledToggle', True, block=True)
store.put_bool('SafeMode', False, block=True)
store.put_bool('DisableOpenpilotLongitudinal', stock, block=True)
store.remove('AlphaLongitudinalEnabled')
store.remove('VehicleSelection')
environment = {k: v for k, v in os.environ.items() if k not in ('FINGERPRINT', 'SKIP_FW_QUERY', 'DISABLE_FW_CACHE')}
with (mock.patch.dict(os.environ, environment, clear=True),
mock.patch.object(car_helpers, 'get_vin', return_value=(0, offset, vin)),
mock.patch.object(car_helpers, 'get_present_ecus', return_value=set()),
mock.patch.object(car_helpers, 'get_fw_versions_ordered', return_value=fw)):
# Actual startup owner is the same pre-create hook used by Card's public acquisition path.
from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences
snapshot = VehicleStartupPreferences.read(store, enabled=True)
ci = car_helpers.get_car(receive, lambda *args: None, lambda _: None, False, False,
forced_candidate=identity if fixed else None,
pre_create_hook=lambda cp, candidate, fingerprints, car_fw: snapshot.prepare(cp, fingerprints=fingerprints))
cp = ci.CP
assert cp.carFingerprint == identity
assert cp.fingerprintSource == (structs.CarParams.FingerprintSource.fixed if fixed else structs.CarParams.FingerprintSource.fw)
assert cp.openpilotLongitudinalControl != stock and cp.pcmCruise == stock
card = Car(CI=ci, RI=ci.RadarInterface(cp))
try:
assert card.aol_qualified and card.CP.alternativeExperience == 32
assert card.CP.safetyConfigs[-1].safetyModel == structs.CarParams.SafetyModel.volkswagenMeb
assert len(card.CP.safetyConfigs) == (2 if offset else 1)
with structs.CarParams.from_bytes(store.get('CarParams')) as published:
assert published.to_dict() == card.CP.to_dict()
finally:
card.vehicle_startup.close()
@@ -0,0 +1,201 @@
"""Saved MEB choices, exact CP/native profiles and independent longitudinal output."""
from types import SimpleNamespace
from pathlib import Path
import pytest
from opendbc.car import structs
from opendbc.car.volkswagen.interface import CarInterface
from opendbc.car.volkswagen.meb_modes import MEB_CARS, base_word, profile_word, qualified, prepare_stock
from opendbc.car.volkswagen.mebcan import MebLongStateMachine
from opendbc.car.volkswagen.values import CarControllerParams
from openpilot.common.params import Params
from openpilot.common.prefix import OpenpilotPrefix
from openpilot.starpilot.car.volkswagen.preferences import stock_long_requested
from openpilot.starpilot.aol.vehicle import native_matches_cp, policy_for
from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences
def params(identity, *, gateway=True, offset=0, fixed=False, alt=False):
fp = {bus: {} for bus in range(8)}
fp[offset][0x25D] = 8
if gateway:
fp[offset + 1][0x13D] = 32
if alt:
fp[offset][0x3DC] = 8
cp = CarInterface.get_params(identity, fp, [], False, False, False)
cp.fingerprintSource = structs.CarParams.FingerprintSource.fixed if fixed else structs.CarParams.FingerprintSource.fw
return cp
@pytest.mark.parametrize('identity', sorted(MEB_CARS))
@pytest.mark.parametrize('offset,fixed', ((0, False), (0, True), (4, False), (4, True)))
@pytest.mark.parametrize('stock', (False, True))
@pytest.mark.parametrize('alt', (False, True))
def test_exact_gateway_configuration_and_camera_boundary(identity, offset, fixed, stock, alt):
cp = params(identity, offset=offset, fixed=fixed, alt=alt)
assert cp.openpilotLongitudinalControl and not cp.pcmCruise
prepare_stock(cp, stock)
assert cp.pcmCruise == stock and cp.openpilotLongitudinalControl != stock
assert qualified(cp) and cp.safetyConfigs[-1].safetyParam == base_word(cp)
policy = policy_for(cp)
assert policy.intent_supported
cp.safetyConfigs[-1].safetyParam |= policy.safety_param_addition
cp.alternativeExperience |= policy.alternative_experience_addition
assert cp.safetyConfigs[-1].safetyParam == profile_word(cp)
assert native_matches_cp(cp, 34, profile_word(cp))
assert not native_matches_cp(cp, 15, profile_word(cp))
if offset:
assert cp.safetyConfigs[0].safetyParam == 0
camera = params(identity, gateway=False, offset=offset, fixed=fixed, alt=alt)
before = camera.to_dict()
prepare_stock(camera, True)
assert camera.to_dict() == before and camera.dashcamOnly
assert not qualified(camera) and not policy_for(camera).intent_supported
@pytest.mark.parametrize('saved,expected', ((None, False), (b'1', False), (b'0', True), (b'', True), (b'broken', True)))
@pytest.mark.parametrize('disable', (False, True))
def test_actual_saved_alpha_off_absence_and_disable_precedence(saved, expected, disable):
with OpenpilotPrefix():
store = Params()
store.remove('AlphaLongitudinalEnabled')
if saved is not None:
Path(store.get_param_path('AlphaLongitudinalEnabled')).write_bytes(saved)
store.put_bool('DisableOpenpilotLongitudinal', disable, block=True)
assert stock_long_requested(store) == expected
snapshot = VehicleStartupPreferences.read(store, enabled=True)
cp = params(sorted(MEB_CARS)[0])
snapshot.prepare(cp)
snapshot.finalize(cp)
assert cp.pcmCruise == (disable or expected)
assert cp.openpilotLongitudinalControl != (disable or expected)
path = Path(store.get_param_path('AlphaLongitudinalEnabled'))
assert (path.read_bytes() if path.exists() else None) == saved
def test_unreadable_alpha_is_stock_and_does_not_rewrite():
class Unreadable:
def get(self, *args, **kwargs):
raise OSError('unreadable fixture')
def get_param_path(self, *args):
raise OSError('unreadable fixture')
assert stock_long_requested(Unreadable())
def test_marked_lateral_only_uses_inactive_accel_and_standby_status():
cp = params(sorted(MEB_CARS)[0])
policy = policy_for(cp)
cp.alternativeExperience = 32
cp.safetyConfigs[-1].safetyParam |= policy.safety_param_addition
machine = MebLongStateMachine(cp, CarControllerParams(cp))
state = SimpleNamespace(out=structs.CarState(canValid=True, vEgo=20.),
esp_standstill=False, esp_hold_confirmation=False)
state.out.cruiseState.available = True
command = structs.CarControl(enabled=True, latActive=True, longActive=False)
command.actuators.longControlState = structs.CarControl.Actuators.LongControlState.pid
accel, status, hold, braking, launching = machine.update(state, command, .5)
assert accel == machine.CCP.ACCEL_INACTIVE
assert status == machine.acc_status_vals['ACC_STANDBY']
assert hold == machine.acc_hold_type_vals['KEINE_ANFORDERUNG'] and not braking and not launching
@pytest.mark.parametrize('pedal', ('gasPressed', 'brakePressed'))
def test_marked_long_withdraws_immediately_on_current_pedal_before_native_status(pedal):
cp = params(sorted(MEB_CARS)[0])
policy = policy_for(cp)
cp.alternativeExperience = 32
cp.safetyConfigs[-1].safetyParam |= policy.safety_param_addition
machine = MebLongStateMachine(cp, CarControllerParams(cp))
state = SimpleNamespace(out=structs.CarState(canValid=True, vEgo=20.),
esp_standstill=False, esp_hold_confirmation=False)
state.out.cruiseState.available = True
setattr(state.out, pedal, True)
command = structs.CarControl(enabled=True, latActive=True, longActive=True)
command.actuators.longControlState = structs.CarControl.Actuators.LongControlState.pid
accel, status, hold, braking, launching = machine.update(state, command, .5)
assert not machine._long_active(state, command)
assert accel == machine.CCP.ACCEL_INACTIVE and status == machine.acc_status_vals['ACC_STANDBY']
assert hold == machine.acc_hold_type_vals['KEINE_ANFORDERUNG'] and not braking and not launching
def test_gateway_ui_saved_off_and_parked_cas_guards():
from openpilot.starpilot.ui.feature_settings_owner import FeatureSettingsOwner
from openpilot.starpilot.ui.feature_settings_state import FeatureSettingsRequest
with OpenpilotPrefix():
store = Params()
store.put_bool('SafeMode', False, block=True)
cp = params(sorted(MEB_CARS)[0])
allowed = [True]
owner = FeatureSettingsOwner(store, lambda group: allowed[0] and group == 'parked_preferences',
vehicle_fingerprint=lambda: cp.carFingerprint, vehicle_params=lambda: cp)
capability = owner._bolt_disable_capability()
assert capability is not None
request = FeatureSettingsRequest('DisableOpenpilotLongitudinal', None, 'On', confirmation=True,
vehicle_fingerprint=cp.carFingerprint, capability=capability)
assert owner.apply(request)
assert not owner.apply(request) # Original source bytes are stale after the successful commit.
request = FeatureSettingsRequest('DisableOpenpilotLongitudinal', b'1', 'Off', confirmation=True,
vehicle_fingerprint=cp.carFingerprint, capability=capability)
allowed[0] = False
assert not owner.apply(request)
allowed[0] = True
store.put_bool('AlphaLongitudinalEnabled', False, block=True)
assert not owner.apply(request) # Cannot override explicit speed-control OFF from the other driving setting.
store.put_bool('AlphaLongitudinalEnabled', True, block=True)
assert owner.apply(request)
@pytest.mark.parametrize('identity', ('VOLKSWAGEN_ID4_MK1', 'VOLKSWAGEN_ID4_MK2'))
def test_temporary_eps_admission_keeps_exact_profile_and_physical_disarms(identity):
from openpilot.starpilot.car.volkswagen.aol import temporary_eps_restriction
cp = params(identity)
policy = policy_for(cp)
cp.alternativeExperience = 32
cp.safetyConfigs[-1].safetyParam |= policy.safety_param_addition
state = structs.CarState(canValid=True, steerFaultTemporary=True, gearShifter=structs.CarState.GearShifter.drive)
state.cruiseState.available = True
assert temporary_eps_restriction(cp, state)
for field in ('steerFaultPermanent', 'brakePressed', 'gasPressed', 'regenBraking', 'doorOpen', 'seatbeltUnlatched',
'accFaulted', 'vehicleSensorsInvalid', 'steeringDisengage', 'canTimeout'):
setattr(state, field, True)
assert not temporary_eps_restriction(cp, state), field
setattr(state, field, False)
for brand in ('honda', 'hyundai', 'gm', 'toyota', 'ford', 'subaru', 'chrysler', 'nissan', 'mazda', 'rivian', 'tesla'):
cp.brand = brand
assert not temporary_eps_restriction(cp, state), brand
cp.brand = 'volkswagen'
cp.alternativeExperience = 0
assert not temporary_eps_restriction(cp, state)
@pytest.mark.parametrize('identity', ('VOLKSWAGEN_ID4_MK1', 'VOLKSWAGEN_ID4_MK2'))
@pytest.mark.parametrize('disarm', ('current_native', 'permanent', 'host_fault', 'main_off', 'can_invalid'))
def test_recovered_eps_ignores_only_prior_denial_and_never_a_current_disarm(identity, disarm):
from openpilot.starpilot.aol.intent import AolSettings
from openpilot.starpilot.car.volkswagen.aol import VolkswagenStockCardIntent
cp = params(identity)
policy = policy_for(cp)
cp.alternativeExperience = 32
cp.safetyConfigs[-1].safetyParam |= policy.safety_param_addition
owner = VolkswagenStockCardIntent(cp, AolSettings(True, 5., 0, 0, (0, 0, 0), (0, 0, 0)))
state = structs.CarState(canValid=True, vEgo=20., gearShifter=structs.CarState.GearShifter.drive)
state.cruiseState.available = True
owner.update(state, now_ns=1_000_000_000, fault_active=False)
assert owner.allowed_latch
state.steerFaultTemporary = True
owner.update(state, now_ns=1_200_000_000, fault_active=False)
assert owner.allowed_latch
state.steerFaultTemporary = False
owner.update(state, now_ns=1_250_000_000, native_rejection_ns=1_200_000_000, fault_active=False)
assert owner.allowed_latch and owner._native_rejection_ns == 0
rejection = 1_200_000_000
if disarm == 'current_native':
rejection = 1_260_000_000
elif disarm == 'permanent':
state.steerFaultPermanent = True
elif disarm == 'main_off':
state.cruiseState.available = False
elif disarm == 'can_invalid':
state.canValid = False
owner.update(state, now_ns=1_260_000_000, native_rejection_ns=rejection, fault_active=disarm == 'host_fault')
assert not owner.allowed_latch
@@ -0,0 +1,64 @@
"""Actual preference/factory/parser/controller and saved OFF matrix for all nine MEBs."""
import pytest
from opendbc.car import Bus, structs
from opendbc.can import CANPacker, CANParser
from opendbc.car.volkswagen.interface import CarInterface
from opendbc.car.volkswagen.meb_modes import MEB_CARS, base_word, qualified
from opendbc.car.volkswagen.values import DBC, CanBus
from openpilot.common.params import Params
from openpilot.common.prefix import OpenpilotPrefix
from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences
@pytest.mark.parametrize('identity', sorted(MEB_CARS))
@pytest.mark.parametrize('disabled', (False, True))
@pytest.mark.parametrize('alt_gear', (False, True))
def test_actual_saved_off_keeps_stock_acc_and_the_selected_curvature_controller(identity, disabled, alt_gear):
fp = {bus: {} for bus in range(8)}
fp[1][0x13D] = 32
if alt_gear:
fp[0][0x3DC] = 8
cp = CarInterface.get_params(identity, fp, [], False, False, False)
with OpenpilotPrefix():
params = Params()
params.put_bool('OpenpilotEnabledToggle', True, block=True)
params.put_bool('SafeMode', False, block=True)
params.put_bool('DisableOpenpilotLongitudinal', disabled, block=True)
preferences = VehicleStartupPreferences.read(params, enabled=True)
preferences.prepare(cp, fingerprints=fp)
ci = CarInterface(cp)
preferences.finalize(cp)
assert cp.openpilotLongitudinalControl == (not disabled)
assert cp.pcmCruise == disabled
assert cp.safetyConfigs[-1].safetyParam == base_word(cp)
assert qualified(cp)
ci.update([])
packer = CANPacker(DBC[identity][Bus.pt])
can = CanBus(cp)
messages = [packer.make_can_msg('Getriebe_11', can.pt, {'GE_Fahrstufe': 8}),
packer.make_can_msg('Gateway_73', can.pt, {'GE_Fahrstufe': 8}),
packer.make_can_msg('Motor_51', can.pt, {'TSK_Status': 3}),
packer.make_can_msg('Motor_54', can.pt, {'Engine_On': 1}),
packer.make_can_msg('QFK_01', can.pt, {'LatCon_HCA_Status': 2}),
packer.make_can_msg('ACC_19', can.ext, {'ACC_Wunschgeschw_02': 72, 'ACC_Limiter_Mode': 0})]
ci.can_parsers[Bus.pt].update([(1_000_000_000, messages)])
ci.can_parsers[Bus.cam].update([(1_000_000_000, messages)])
state = ci.CS.update(ci.can_parsers)
assert state.gearShifter == structs.CarState.GearShifter.drive
assert state.cruiseState.available and state.cruiseState.enabled
if disabled:
assert state.cruiseState.speed == pytest.approx(20., abs=.1)
ci.CS.out = state
command = structs.CarControl()
command.cruiseControl.cancel = True
_, sent = ci.apply(command.as_reader(), 1_010_000_000)
ids = {addr for addr, _, _ in sent}
assert (0x14D in ids) == (not disabled)
if disabled:
assert not ids & {0x14D, 0x300, 0x26B}
assert 0x12B in ids # Actual restored stock Cancel command factory.
observe = CANParser(DBC[identity][Bus.pt], [('GRA_ACC_01', 0)], can.ext)
observe.update([(1_010_000_000, sent)])
assert observe.vl['GRA_ACC_01']['GRA_Abbrechen'] == 1
assert observe.vl['GRA_ACC_01']['COUNTER'] == (ci.CS.gra_stock_values['COUNTER'] + 1) % 16
assert all(bus == can.ext for addr, _, bus in sent if addr == 0x12B)
@@ -45,8 +45,8 @@ def test_normal_golf_discovery_and_card_stock_aol_owner(manual, gateway, alpha,
'subAddress': None,
'fwVersion': 'f1873551303930373536374720f18930333930f182013031'}],
'origin': 'Synthetic ECU bytes from configuration-matrix.json; actual matcher must verify uniqueness.'}
fw = [structs.CarParams.CarFw(ecu=getattr(structs.CarParams.Ecu, r['ecu']),
address=r['address'], subAddress=r['subAddress'] or 0, fwVersion=bytes.fromhex(r['fwVersion']),
fw = [structs.CarParams.CarFw(ecu=getattr(structs.CarParams.Ecu, str(r['ecu'])),
address=r['address'], subAddress=r['subAddress'] or 0, fwVersion=bytes.fromhex(str(r['fwVersion'])),
brand='volkswagen', bus=offset) for r in fixture['firmware']]
exact, candidates = match_fw_to_car(fw, fixture['vin'], log=False)
assert exact and candidates == {CAR.VOLKSWAGEN_GOLF_MK7}, (exact, candidates)
@@ -23,6 +23,8 @@
#define VOLKSWAGEN_MEB_ACC_AKTIV_REGELT 3U
#define VOLKSWAGEN_MEB_ACC_OVERRIDE 4U
#include "opendbc/safety/modes/volkswagen_meb_aol.h"
static bool volkswagen_meb_alt_crc = false;
#define VOLKSWAGEN_MEB_COMMON_RX_CHECKS \
@@ -96,6 +98,14 @@ static uint32_t volkswagen_meb_alt_crc_compute(const CANPacket_t *msg) {
}
static safety_config volkswagen_meb_init(uint16_t param) {
static const CanMsg VOLKSWAGEN_MEB_STOCK_TX_MSGS[] = {
{MSG_HCA_03, 0, 24, .check_relay = true},
{MSG_GRA_ACC_01, 0, 8, .check_relay = false},
{MSG_GRA_ACC_01, 2, 8, .check_relay = false},
{MSG_LDW_02, 0, 8, .check_relay = true},
{MSG_KLR_01, 0, 8, .check_relay = false},
{MSG_KLR_01, 2, 8, .check_relay = true},
};
static const CanMsg VOLKSWAGEN_MEB_TX_MSGS[] = {
{MSG_HCA_03, 0, 24, .check_relay = true},
{MSG_LDW_02, 0, 8, .check_relay = true},
@@ -109,6 +119,22 @@ static safety_config volkswagen_meb_init(uint16_t param) {
volkswagen_common_init();
const uint16_t FLAG_VOLKSWAGEN_MEB_ALT_CRC = 2;
volkswagen_meb_alt_crc = GET_FLAG(param, FLAG_VOLKSWAGEN_MEB_ALT_CRC);
volkswagen_meb_stock_acc = GET_FLAG(param, 4U);
vw_meb_aol_reset();
const bool aol_word = (param == 8U) || (param == 10U) || (param == 12U) || (param == 14U) ||
(param == 24U) || (param == 26U) || (param == 28U) || (param == 30U);
vw_meb_aol_enabled = ((unsigned int)alternative_experience == 32U) && aol_word;
vw_meb_aol_alt_gear = GET_FLAG(param, 16U);
if (vw_meb_aol_enabled) {
static const AolSafetyPolicy policy = {
.reset = vw_meb_aol_reset,
.host_request = vw_meb_aol_request,
.request_mask = vw_meb_aol_request_mask,
.permission_mask = vw_meb_aol_permission,
.rx_invalid = vw_meb_aol_clear,
};
aol_policy = &policy;
}
safety_config ret;
if (volkswagen_meb_alt_crc) {
@@ -128,6 +154,15 @@ static safety_config volkswagen_meb_init(uint16_t param) {
ret = BUILD_SAFETY_CFG(volkswagen_meb_rx_checks, VOLKSWAGEN_MEB_TX_MSGS);
}
if (volkswagen_meb_stock_acc) {
ret.tx_msgs = VOLKSWAGEN_MEB_STOCK_TX_MSGS;
ret.tx_msgs_len = sizeof(VOLKSWAGEN_MEB_STOCK_TX_MSGS) / sizeof(CanMsg);
}
if ((((param & ~6U) != 0U) && !vw_meb_aol_enabled) ||
(((unsigned int)alternative_experience == 32U) && !vw_meb_aol_enabled)) {
ret.tx_msgs = NULL;
ret.tx_msgs_len = 0;
}
return ret;
}
@@ -164,6 +199,9 @@ static void volkswagen_meb_rx_hook(const CANPacket_t *msg) {
bool cruise_engaged = (acc_status == 3) || (acc_status == 4) || (acc_status == 5);
acc_main_on = cruise_engaged || (acc_status == 2);
if (volkswagen_meb_stock_acc) {
pcm_cruise_check(cruise_engaged);
}
if (!acc_main_on) {
controls_allowed = false;
}
@@ -178,7 +216,8 @@ static void volkswagen_meb_rx_hook(const CANPacket_t *msg) {
// Signal: GRA_ACC_01.GRA_Tip_Wiederaufnahme
bool set_button = GET_BIT(msg, 16U);
bool resume_button = GET_BIT(msg, 19U);
if ((volkswagen_set_button_prev && !set_button) || (volkswagen_resume_button_prev && !resume_button)) {
if (!volkswagen_meb_stock_acc &&
((volkswagen_set_button_prev && !set_button) || (volkswagen_resume_button_prev && !resume_button))) {
controls_allowed = acc_main_on;
}
volkswagen_set_button_prev = set_button;
@@ -193,6 +232,7 @@ static void volkswagen_meb_rx_hook(const CANPacket_t *msg) {
if (msg_matches(msg, MSG_MOTOR_14, 0U)) {
brake_pressed = GET_BIT(msg, 28U);
}
vw_meb_aol_rx(msg);
}
static bool volkswagen_meb_tx_hook(const CANPacket_t *msg) {
@@ -210,7 +250,8 @@ static bool volkswagen_meb_tx_hook(const CANPacket_t *msg) {
// Signal: ACC_18.ACC_Sollbeschleunigung_02 (acceleration in m/s2, scale 0.005, offset -7.22)
int desired_accel = ((((msg->data[4] & 0x7U) << 8) | msg->data[3]) * 5U) - 7220U;
// MEB inactive accel is 3.01, but we also need to send 0.0 for gas override
bool accel_override = controls_allowed && (desired_accel == 0);
const bool current_long_allowed = vw_meb_aol_enabled ? longitudinal_controls_allowed() : controls_allowed;
bool accel_override = current_long_allowed && (desired_accel == 0);
if (!accel_override && longitudinal_accel_checks(desired_accel, VOLKSWAGEN_MEB_LONG_LIMITS)) {
tx = false;
}
@@ -222,7 +263,7 @@ static bool volkswagen_meb_tx_hook(const CANPacket_t *msg) {
uint8_t hold_type = (msg->data[9] >> 5) & 0x07U;
bool hold_type_allowed = (hold_type == VOLKSWAGEN_MEB_HMS_KEINE_ANFORDERUNG) ||
(hold_type == VOLKSWAGEN_MEB_HMS_LOESEN_UEBER_RAMPE) ||
(controls_allowed && ((hold_type == VOLKSWAGEN_MEB_HMS_HALTEN) ||
(current_long_allowed && ((hold_type == VOLKSWAGEN_MEB_HMS_HALTEN) ||
(hold_type == VOLKSWAGEN_MEB_HMS_ANFAHREN)));
if (!hold_type_allowed) {
tx = false;
@@ -233,7 +274,7 @@ static bool volkswagen_meb_tx_hook(const CANPacket_t *msg) {
// These carry the same drive off and hold requests as the hold type, so they are gated the same way
bool acc_anfahren = GET_BIT(msg, 56U);
bool acc_anhalten = GET_BIT(msg, 57U);
if ((acc_anfahren || acc_anhalten) && !controls_allowed) {
if ((acc_anfahren || acc_anhalten) && !current_long_allowed) {
tx = false;
}
@@ -243,7 +284,7 @@ static bool volkswagen_meb_tx_hook(const CANPacket_t *msg) {
uint8_t acc_status = (msg->data[7] >> 4) & 0x07U;
bool acc_status_active = (acc_status == VOLKSWAGEN_MEB_ACC_AKTIV_REGELT) ||
(acc_status == VOLKSWAGEN_MEB_ACC_OVERRIDE);
if (acc_status_active && !controls_allowed) {
if (acc_status_active && !current_long_allowed) {
tx = false;
}
}
@@ -273,6 +314,9 @@ static bool volkswagen_meb_tx_hook(const CANPacket_t *msg) {
}
}
if ((msg->addr == MSG_GRA_ACC_01) && !controls_allowed && ((msg->data[2] & 0x9U) != 0U)) {
tx = false;
}
return tx;
}
@@ -280,6 +324,7 @@ const safety_hooks volkswagen_meb_hooks = {
.init = volkswagen_meb_init,
.rx = volkswagen_meb_rx_hook,
.tx = volkswagen_meb_tx_hook,
.optional_rx = vw_meb_aol_optional_rx,
.get_counter = volkswagen_mxb_get_counter,
.get_checksum = volkswagen_mxb_get_checksum,
.compute_checksum = volkswagen_meb_alt_crc_compute,
@@ -0,0 +1,196 @@
#pragma once
static bool volkswagen_meb_stock_acc = false;
static bool vw_meb_aol_enabled = false;
static bool vw_meb_aol_alt_gear = false;
static bool vw_meb_aol_initial_main = true;
static bool vw_meb_aol_token = false;
static bool vw_meb_aol_long_armed = false;
static bool vw_meb_aol_cruise_engaged = false;
static bool vw_meb_aol_main_neutral = false;
static bool vw_meb_aol_requested = false;
static bool vw_meb_aol_gear_drive = false;
static bool vw_meb_aol_eps_ready = false;
static bool vw_meb_aol_eps_initialized = false;
static uint8_t vw_meb_aol_seen = 0U;
static uint8_t vw_meb_aol_buttons = 0U;
static bool vw_meb_aol_counter_seen = false;
static uint8_t vw_meb_aol_gear_counter = 0U;
static uint32_t vw_meb_aol_main_ts = 0U;
static uint32_t vw_meb_aol_gear_ts = 0U;
static uint32_t vw_meb_aol_eps_ts = 0U;
static uint32_t vw_meb_aol_eps_init_ts = 0U;
static void vw_meb_aol_clear(void) {
vw_meb_aol_token = false;
vw_meb_aol_long_armed = false;
vw_meb_aol_main_neutral = false;
vw_meb_aol_buttons = 0U;
}
static void vw_meb_aol_reset(void) {
vw_meb_aol_clear();
vw_meb_aol_enabled = false;
vw_meb_aol_alt_gear = false;
vw_meb_aol_initial_main = true;
vw_meb_aol_cruise_engaged = false;
vw_meb_aol_requested = false;
vw_meb_aol_gear_drive = false;
vw_meb_aol_eps_ready = false;
vw_meb_aol_eps_initialized = false;
vw_meb_aol_seen = 0U;
vw_meb_aol_counter_seen = false;
vw_meb_aol_gear_counter = 0U;
vw_meb_aol_main_ts = 0U;
vw_meb_aol_gear_ts = 0U;
vw_meb_aol_eps_ts = 0U;
vw_meb_aol_eps_init_ts = 0U;
}
static bool vw_meb_aol_current(void) {
const uint32_t now = microsecond_timer_get();
return (vw_meb_aol_seen == 7U) &&
(safety_get_ts_elapsed(now, vw_meb_aol_main_ts) <= 300000U) &&
(safety_get_ts_elapsed(now, vw_meb_aol_gear_ts) <= 300000U) &&
(safety_get_ts_elapsed(now, vw_meb_aol_eps_ts) <= 100000U);
}
static void vw_meb_aol_request(uint8_t mask) {
vw_meb_aol_requested |= mask != 0U;
}
static uint8_t vw_meb_aol_request_mask(void) {
const bool expired = vw_meb_aol_requested &&
(safety_get_ts_elapsed(microsecond_timer_get(), aol_host_request_ts) > AOL_HOST_REQUEST_TIMEOUT_US);
uint8_t request = 0U;
if (!vw_meb_aol_enabled || relay_malfunction || safety_rx_checks_invalid || !heartbeat_engaged ||
!aol_rx_healthy() || !vw_meb_aol_current() || expired) {
vw_meb_aol_clear();
} else {
request = aol_host_axis_mask;
}
return request;
}
static uint8_t vw_meb_aol_permission(void) {
const uint8_t request = vw_meb_aol_request_mask();
uint8_t permission = 0U;
if (acc_main_on && vw_meb_aol_gear_drive && vw_meb_aol_eps_ready && vw_meb_aol_token) {
permission = request & 1U;
}
if (!volkswagen_meb_stock_acc && vw_meb_aol_long_armed && controls_allowed && acc_main_on &&
vw_meb_aol_gear_drive && !brake_pressed && !gas_pressed) {
permission |= request & 2U;
}
return permission;
}
static void vw_meb_aol_optional_rx(const CANPacket_t *msg) {
if (vw_meb_aol_enabled) {
unsigned int gear = 0U;
bool accepted = false;
if (vw_meb_aol_alt_gear && msg_matches(msg, 0x3DCU, 0U, 8U)) {
gear = msg->data[5] & 15U;
accepted = true; // Gateway_73 has no checksum/counter in the current DBC.
} else if (!vw_meb_aol_alt_gear && msg_matches(msg, 0xADU, 0U, 8U)) {
const uint8_t counter = volkswagen_mxb_get_counter(msg);
const uint8_t salt[16] = {0x3FU, 0x69U, 0x39U, 0xDCU, 0x94U, 0xF9U, 0x14U, 0x64U,
0xD8U, 0x6AU, 0x34U, 0xCEU, 0xA2U, 0x55U, 0xB5U, 0x2CU};
const bool checksum = volkswagen_compute_crc(msg, 8, salt[counter]) == msg->data[0];
const bool pair = vw_meb_aol_counter_seen &&
(safety_get_ts_elapsed(microsecond_timer_get(), vw_meb_aol_gear_ts) <= 300000U) &&
(counter == ((vw_meb_aol_gear_counter + 1U) & 15U));
vw_meb_aol_counter_seen = checksum;
vw_meb_aol_gear_counter = counter;
if (checksum) {
vw_meb_aol_gear_ts = microsecond_timer_get();
}
if (checksum && pair) {
gear = (msg->data[5] >> 2) & 15U;
accepted = true;
} else {
vw_meb_aol_seen &= 5U;
vw_meb_aol_gear_drive = false;
}
} else {
}
if (accepted) {
vw_meb_aol_seen |= 2U;
vw_meb_aol_gear_ts = microsecond_timer_get();
vw_meb_aol_gear_drive = (gear == 8U) || (gear == 9U) || (gear == 10U) || (gear == 13U) || (gear == 14U);
}
if (!vw_meb_aol_gear_drive) {
vw_meb_aol_long_armed = false;
}
}
}
static void vw_meb_aol_rx(const CANPacket_t *msg) {
if (vw_meb_aol_enabled) {
if (brake_pressed || !vw_meb_aol_gear_drive) {
vw_meb_aol_long_armed = false;
}
if (msg_matches(msg, MSG_QFK_01, 0U, 32U)) {
const unsigned int eps = (msg->data[2] >> 4) & 7U;
if ((vw_meb_aol_seen & 4U) == 0U) {
vw_meb_aol_eps_init_ts = microsecond_timer_get();
}
vw_meb_aol_seen |= 4U;
vw_meb_aol_eps_ts = microsecond_timer_get();
vw_meb_aol_eps_ready = (eps == 2U) || (eps == 4U);
// Match CarState's existing 600-frame initialization window; no permission is granted while EPS is unavailable.
vw_meb_aol_eps_initialized |= (eps == 0U) || (eps == 2U) || (eps == 4U) ||
(safety_get_ts_elapsed(microsecond_timer_get(), vw_meb_aol_eps_init_ts) > 6000000U);
if (((eps == 0U) && vw_meb_aol_gear_drive) ||
(vw_meb_aol_eps_initialized && ((eps == 3U) || (eps == 6U)))) {
vw_meb_aol_initial_main = false;
vw_meb_aol_clear();
}
} else if (msg_matches(msg, MSG_Motor_51, 0U)) {
const unsigned int status = msg->data[11] & 7U;
const bool engaged = (status == 3U) || (status == 4U) || (status == 5U);
const bool edge = engaged && !vw_meb_aol_cruise_engaged;
vw_meb_aol_cruise_engaged = engaged;
vw_meb_aol_seen |= 1U;
vw_meb_aol_main_ts = microsecond_timer_get();
if (!heartbeat_engaged || !aol_rx_healthy() || !vw_meb_aol_current() || safety_rx_checks_invalid) {
vw_meb_aol_clear();
} else if (!acc_main_on) {
vw_meb_aol_initial_main = false;
vw_meb_aol_clear();
vw_meb_aol_main_neutral = status < 2U;
} else if (vw_meb_aol_gear_drive && vw_meb_aol_eps_ready && !brake_pressed &&
(vw_meb_aol_initial_main || vw_meb_aol_main_neutral ||
(volkswagen_meb_stock_acc && edge && controls_allowed))) {
vw_meb_aol_initial_main = false;
vw_meb_aol_main_neutral = false;
vw_meb_aol_token = true;
} else {
}
} else if (msg_matches(msg, MSG_GRA_ACC_01, 0U, 8U)) {
if (GET_BIT(msg, 13U)) {
vw_meb_aol_initial_main = false;
vw_meb_aol_clear();
} else if (!volkswagen_meb_stock_acc) {
const bool healthy = heartbeat_engaged && aol_rx_healthy() && vw_meb_aol_current() && !safety_rx_checks_invalid;
const uint8_t buttons = (GET_BIT(msg, 16U) ? 1U : 0U) | (GET_BIT(msg, 19U) ? 2U : 0U);
if (!healthy) {
vw_meb_aol_clear();
} else if (buttons == 0U) {
if (((vw_meb_aol_buttons & 3U) != 0U) && controls_allowed && acc_main_on &&
vw_meb_aol_gear_drive && vw_meb_aol_eps_ready && !brake_pressed) {
vw_meb_aol_token = true;
vw_meb_aol_long_armed = true;
}
vw_meb_aol_buttons = 4U;
} else if ((vw_meb_aol_buttons & 4U) != 0U) {
vw_meb_aol_buttons = 4U | buttons;
} else {
}
} else {
}
} else {
}
}
}
@@ -0,0 +1,333 @@
"""Actual MEB native, CAN-FD CRC generations, stock/off and independent axes."""
import unittest
import pytest
from opendbc.car import structs
from opendbc.can.dbc import DBC
from opendbc.safety.tests.common import CANPackerSafety
from opendbc.safety.tests.libsafety import libsafety_py
class _MebOwner:
def __init__(self, gen2, stock, alt_gear):
self.safety = libsafety_py.libsafety
self.safety.init_tests()
self.safety.set_alternative_experience(32)
self.word = 8 | (2 if gen2 else 0) | (4 if stock else 0) | (16 if alt_gear else 0)
self.safety.set_safety_hooks(34, self.word)
self.safety.set_aol_test_heartbeat(True)
self.packer = CANPackerSafety('vw_meb_2024_generated' if gen2 else 'vw_meb_generated')
self.dbc = DBC(self.packer.dbc_name) if hasattr(self.packer, 'dbc_name') else DBC('vw_meb_2024_generated' if gen2 else 'vw_meb_generated')
self.time = 1_000_000
self.stock, self.alt_gear = stock, alt_gear
def packet(self, name, fields, bus=0):
return self.packer.make_can_msg_safety(name, bus, fields)
def feed(self, engaged=False, brake=False, gear=8, eps=2, main=True, set_button=False, cancel=False, omit=None, moving=True):
self.time += 10_000
self.safety.set_timer(self.time)
data = [('Gateway_73' if self.alt_gear else 'Getriebe_11', {'GE_Fahrstufe': gear}),
('LH_EPS_03', {'EPS_Lenkmoment': 0}), ('Motor_14', {'MO_Fahrer_bremst': int(brake)}),
('GRA_ACC_01', {'GRA_Tip_Setzen': int(set_button), 'GRA_Abbrechen': int(cancel)}),
('QFK_01', {'LatCon_HCA_Status': eps, 'Curvature': 0}),
('ESP_21', {'ESP_v_Signal': 72 if moving else 0}),
('ESC_51', {name + '_Radgeschw': 72 if moving else 0 for name in ('VL', 'VR', 'HL', 'HR')}),
('Motor_51', {'TSK_Status': 3 if engaged else 2 if main else 0, 'Accel_Pedal_Pressure': 0})]
for name, fields in data:
if name != omit:
self.safety.safety_rx_hook(self.packet(name, fields))
self.safety.safety_tick()
self.safety.aol_set_host_request(3)
def close(self):
self.safety.init_tests()
self.safety.set_alternative_experience(0)
self.safety.set_safety_hooks(structs.CarParams.SafetyModel.noOutput, 0)
@pytest.mark.parametrize('gen2', (False, True))
@pytest.mark.parametrize('stock', (False, True))
@pytest.mark.parametrize('alt_gear', (False, True))
def test_actual_meb_native_full_engagement_brake_reverse_and_long_off(gen2, stock, alt_gear):
owner = _MebOwner(gen2, stock, alt_gear)
try:
for _ in range(15):
owner.feed()
assert owner.safety.aol_get_permission_mask() == 1 # Current cold MAIN arms LAT only.
for address in (0x14D, 0x300, 0x26B):
assert owner.safety.safety_fwd_hook(2, address) == (0 if stock else -1)
if stock:
owner.feed(engaged=True)
else:
owner.feed(set_button=True)
owner.feed(set_button=False)
assert owner.safety.aol_get_permission_mask() == (1 if stock else 3)
assert owner.safety.safety_tx_hook(owner.packet('HCA_03', {'Curvature': 0, 'Power': 4, 'RequestStatus': 4, 'HighSendRate': 1}))
active = {'ACC_Sollbeschleunigung_02': .5, 'ACC_Status_ACC': 3, 'ACC_Anforderung_HMS': 0}
assert bool(owner.safety.safety_tx_hook(owner.packet('ACC_18', active))) == (not stock)
owner.feed(brake=True)
assert owner.safety.aol_get_permission_mask() == 1
assert not owner.safety.get_controls_allowed()
assert not owner.safety.safety_tx_hook(owner.packet('ACC_18', active))
owner.feed()
assert owner.safety.aol_get_permission_mask() == 1
for gear in (6, 7):
owner.feed(gear=gear)
assert owner.safety.aol_get_permission_mask() == 0
owner.feed()
assert owner.safety.aol_get_permission_mask() == 1
owner.feed(eps=7)
assert owner.safety.aol_get_permission_mask() == 0
owner.feed()
assert owner.safety.aol_get_permission_mask() == 1
owner.feed(eps=3)
assert owner.safety.aol_get_permission_mask() == 0
owner.feed()
assert owner.safety.aol_get_permission_mask() == 0
finally:
owner.close()
@pytest.mark.parametrize('gen2', (False, True))
def test_meb_gear_crc_counter_and_wrong_bus_withdraw(gen2):
owner = _MebOwner(gen2, False, False)
try:
for _ in range(15):
owner.feed()
owner.feed(set_button=True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 3
frame = owner.packet('Getriebe_11', {'GE_Fahrstufe': 8})
frame[0].data[0] ^= 255
owner.safety.safety_rx_hook(frame)
assert owner.safety.aol_get_permission_mask() == 0
for _ in range(40):
owner.feed(omit='Getriebe_11')
owner.safety.safety_rx_hook(owner.packet('Getriebe_11', {'GE_Fahrstufe': 8}, 1))
assert owner.safety.aol_get_permission_mask() == 0
owner.safety.set_aol_test_heartbeat(False)
assert owner.safety.aol_get_permission_mask() == 0
finally:
owner.close()
@pytest.mark.parametrize('gen2', (False, True))
@pytest.mark.parametrize('stock', (False, True))
def test_meb_current_main_never_grants_long_or_recovers_long_automatically(gen2, stock):
owner = _MebOwner(gen2, stock, False)
try:
for _ in range(12):
owner.feed()
assert owner.safety.aol_get_permission_mask() == 1
for active in (
{'ACC_Sollbeschleunigung_02': .5, 'ACC_Status_ACC': 3},
{'ACC_Sollbeschleunigung_02': 0, 'ACC_Status_ACC': 4},
{'ACC_Sollbeschleunigung_02': 3.01, 'ACC_Anforderung_HMS': 1},
{'ACC_Sollbeschleunigung_02': 3.01, 'ACC_Anforderung_HMS': 4},
{'ACC_Sollbeschleunigung_02': 3.01, 'ACC_Anfahren': 1},
{'ACC_Sollbeschleunigung_02': 3.01, 'ACC_Anhalten': 1},
):
assert not owner.safety.safety_tx_hook(owner.packet('ACC_18', active))
owner.feed(set_button=True, engaged=stock)
owner.feed(engaged=stock)
assert owner.safety.aol_get_permission_mask() == (1 if stock else 3)
owner.feed(gear=6)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 1
owner.feed(eps=3)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 0
owner.feed(set_button=True, engaged=stock)
owner.feed(engaged=stock)
assert owner.safety.aol_get_permission_mask() == (1 if stock else 3)
owner.feed(cancel=True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 0
finally:
owner.close()
@pytest.mark.parametrize('gen2', (False, True))
@pytest.mark.parametrize('alt_gear', (False, True))
@pytest.mark.parametrize('source', ('LH_EPS_03', 'Motor_14', 'GRA_ACC_01', 'QFK_01', 'ESP_21', 'ESC_51', 'Motor_51', 'gear'))
def test_meb_each_required_source_lease_withdraws_without_rearming(gen2, alt_gear, source):
owner = _MebOwner(gen2, False, alt_gear)
try:
for _ in range(12):
owner.feed()
owner.feed(set_button=True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 3
omitted = ('Gateway_73' if alt_gear else 'Getriebe_11') if source == 'gear' else source
for _ in range(120):
owner.feed(omit=omitted)
assert owner.safety.aol_get_permission_mask() == 0
for _ in range(12):
owner.feed()
assert owner.safety.aol_get_permission_mask() == 0
owner.feed(set_button=True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 3
finally:
owner.close()
@pytest.mark.parametrize('gen2', (False, True))
def test_meb_permanent_fault_held_button_requires_fresh_press(gen2):
owner = _MebOwner(gen2, False, False)
try:
for _ in range(12):
owner.feed()
owner.feed(set_button=True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 3
owner.feed(eps=3, set_button=True)
owner.feed(set_button=True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 0
owner.feed(set_button=True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 3
owner.safety.set_aol_test_heartbeat(False)
assert owner.safety.aol_get_permission_mask() == 0
owner.safety.set_aol_test_heartbeat(True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 0
finally:
owner.close()
@pytest.mark.parametrize('gen2', (False, True))
def test_meb_main_off_and_counter_replay_cannot_restore_long(gen2):
owner = _MebOwner(gen2, False, False)
try:
for _ in range(12):
owner.feed()
owner.feed(set_button=True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 3
owner.feed(main=False)
assert owner.safety.aol_get_permission_mask() == 0
owner.feed()
assert owner.safety.aol_get_permission_mask() == 1
frame = owner.packet('Getriebe_11', {'GE_Fahrstufe': 8})
owner.safety.safety_rx_hook(frame)
owner.safety.safety_rx_hook(frame)
assert owner.safety.aol_get_permission_mask() == 0
for _ in range(12):
owner.feed()
assert owner.safety.aol_get_permission_mask() == 0
owner.feed(set_button=True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 3
finally:
owner.close()
@pytest.mark.parametrize('gen2', (False, True))
@pytest.mark.parametrize('initial_eps', (1, 3, 6))
def test_meb_initial_eps_unavailability_and_reverse_disabled_follow_real_parser_semantics(gen2, initial_eps):
owner = _MebOwner(gen2, False, False)
try:
for _ in range(15):
owner.feed(eps=initial_eps)
assert owner.safety.aol_get_permission_mask() == 0
owner.feed()
assert owner.safety.aol_get_permission_mask() == 1
owner.feed(gear=6, eps=0)
assert owner.safety.aol_get_permission_mask() == 0
owner.feed()
assert owner.safety.aol_get_permission_mask() == 1
owner.feed(eps=0)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 0 # DISABLED in Drive is permanent.
finally:
owner.close()
@pytest.mark.parametrize('gen2', (False, True))
def test_meb_fault_after_original_initialization_window_cannot_rearm_from_held_main(gen2):
owner = _MebOwner(gen2, False, False)
try:
for _ in range(650):
owner.feed(eps=3)
assert owner.safety.aol_get_permission_mask() == 0
owner.feed()
assert owner.safety.aol_get_permission_mask() == 0
owner.feed(set_button=True)
owner.feed()
assert owner.safety.aol_get_permission_mask() == 3
finally:
owner.close()
@pytest.mark.parametrize('gen2', (False, True))
def test_meb_stop_hold_resume_requires_current_ordinary_long_grant(gen2):
owner = _MebOwner(gen2, False, False)
try:
for _ in range(12):
owner.feed()
owner.feed(set_button=True)
owner.feed()
owner.feed(moving=False)
assert owner.safety.aol_get_permission_mask() == 3
for hold in (1, 4):
frame = owner.packet('ACC_18', {'ACC_Sollbeschleunigung_02': 3.01, 'ACC_Status_ACC': 3,
'ACC_Anforderung_HMS': hold})
assert owner.safety.safety_tx_hook(frame)
assert not owner.safety.safety_tx_hook(owner.packet('ACC_18', {'ACC_Sollbeschleunigung_02': 3.01,
'ACC_Anforderung_HMS': 7}))
owner.feed(brake=True, moving=False)
owner.feed(moving=False)
assert owner.safety.aol_get_permission_mask() == 1
for hold in (1, 4):
assert not owner.safety.safety_tx_hook(owner.packet('ACC_18', {'ACC_Sollbeschleunigung_02': 3.01,
'ACC_Status_ACC': 3, 'ACC_Anforderung_HMS': hold}))
assert owner.safety.safety_tx_hook(owner.packet('ACC_18', {'ACC_Sollbeschleunigung_02': 3.01,
'ACC_Status_ACC': 2, 'ACC_Anforderung_HMS': 5}))
finally:
owner.close()
class TestMebNativeUnittest(unittest.TestCase):
"""Execute the same physical-source assertions in the native unittest coverage gate."""
def test_all_native_profiles(self):
for gen2 in (False, True):
for stock in (False, True):
for alt in (False, True):
test_actual_meb_native_full_engagement_brake_reverse_and_long_off(gen2, stock, alt)
test_meb_current_main_never_grants_long_or_recovers_long_automatically(gen2, stock)
test_meb_gear_crc_counter_and_wrong_bus_withdraw(gen2)
test_meb_permanent_fault_held_button_requires_fresh_press(gen2)
test_meb_main_off_and_counter_replay_cannot_restore_long(gen2)
test_meb_stop_hold_resume_requires_current_ordinary_long_grant(gen2)
test_meb_fault_after_original_initialization_window_cannot_rearm_from_held_main(gen2)
for initial_eps in (1, 3, 6):
test_meb_initial_eps_unavailability_and_reverse_disabled_follow_real_parser_semantics(gen2, initial_eps)
def test_all_native_source_leases(self):
for gen2 in (False, True):
for alt in (False, True):
for source in ('LH_EPS_03', 'Motor_14', 'GRA_ACC_01', 'QFK_01', 'ESP_21', 'ESC_51', 'Motor_51', 'gear'):
test_meb_each_required_source_lease_withdraws_without_rearming(gen2, alt, source)
def test_invalid_profile_and_unarmed_set_resume_are_denied(self):
safety = libsafety_py.libsafety
try:
for experience, word in ((0, 8), (32, 4), (0, 0x8000)):
safety.init_tests()
safety.set_alternative_experience(experience)
safety.set_safety_hooks(structs.CarParams.SafetyModel.volkswagenMeb, word)
self.assertFalse(safety.safety_tx_hook(libsafety_py.make_CANPacket(0x12B, 0, bytes(8))))
for gen2 in (False, True):
owner = _MebOwner(gen2, True, False)
for _ in range(15):
owner.feed()
self.assertFalse(safety.get_controls_allowed())
self.assertFalse(safety.safety_tx_hook(owner.packet('GRA_ACC_01', {'GRA_Tip_Setzen': 1})))
self.assertTrue(safety.safety_tx_hook(owner.packet('GRA_ACC_01', {'GRA_Abbrechen': 1})))
owner.close()
finally:
safety.init_tests()
safety.set_alternative_experience(0)
safety.set_safety_hooks(structs.CarParams.SafetyModel.noOutput, 0)
@@ -10,15 +10,13 @@ from opendbc.safety.tests.libsafety import libsafety_py
@contextmanager
def direct_owner():
owner = stock.TestVolkswagenStockAol()
class DirectOwner(stock.TestVolkswagenStockAol):
def packet(self, name, fields, bus=0):
if name == 'Gateway_73':
return super().packet('Motor_EV_01', {'MO_Waehlpos': 5 if fields['GE_Fahrstufe'] == 8 else 3}, bus)
return super().packet(name, fields, bus)
owner = DirectOwner()
owner.configure(40)
original = owner.packet
def packet(name, fields, bus=0):
if name == 'Gateway_73':
return original('Motor_EV_01', {'MO_Waehlpos': 5 if fields['GE_Fahrstufe'] == 8 else 3}, bus)
return original(name, fields, bus)
owner.packet = packet
try:
yield owner
finally:
Binary file not shown.
+1 -1
View File
@@ -480,7 +480,7 @@ void pandad_run(std::vector<Panda *> &pandas) {
Panda *previous_aol_owner = nullptr;
bool engaged = false;
bool is_onroad = false;
constexpr AolSafetyProfile aol_profiles[] = {HONDA_AOL_PROFILE, HONDA_STOCK_AOL_PROFILE, HONDA_NIDEC_AOL_PROFILE, HYUNDAI_AOL_PROFILE, HYUNDAI_CLASSIC_AOL_PROFILE, HYUNDAI_LEGACY_AOL_PROFILE, GM_AOL_PROFILE, FORD_AOL_PROFILE, MAZDA_AOL_PROFILE, RIVIAN_AOL_PROFILE, CHRYSLER_AOL_PROFILE, TESLA_PREAP_AOL_PROFILE, TESLA_SCREEN_AOL_PROFILE, TOYOTA_AOL_PROFILE, NISSAN_AOL_PROFILE, VOLKSWAGEN_STOCK_AOL_PROFILE, VOLKSWAGEN_LONG_AOL_PROFILE, SUBARU_AOL_PROFILE};
constexpr AolSafetyProfile aol_profiles[] = {HONDA_AOL_PROFILE, HONDA_STOCK_AOL_PROFILE, HONDA_NIDEC_AOL_PROFILE, HYUNDAI_AOL_PROFILE, HYUNDAI_CLASSIC_AOL_PROFILE, HYUNDAI_LEGACY_AOL_PROFILE, GM_AOL_PROFILE, FORD_AOL_PROFILE, MAZDA_AOL_PROFILE, RIVIAN_AOL_PROFILE, CHRYSLER_AOL_PROFILE, TESLA_PREAP_AOL_PROFILE, TESLA_SCREEN_AOL_PROFILE, TOYOTA_AOL_PROFILE, NISSAN_AOL_PROFILE, VOLKSWAGEN_STOCK_AOL_PROFILE, VOLKSWAGEN_LONG_AOL_PROFILE, VOLKSWAGEN_MEB_AOL_PROFILE, SUBARU_AOL_PROFILE};
const AolProfileRegistry aol_registry{aol_profiles, std::size(aol_profiles)};
AolAxisNegotiator aol_negotiator(aol_registry);
// Main loop: receive CAN first, then process lower priority panda and peripheral state.
@@ -28,7 +28,7 @@ constexpr AolSafetyProfile TEST_PROFILES[] = {
{TEST_MODE, test_param, false}, {TEST_ALT_MODE, test_alt_param, false},
};
const AolProfileRegistry TEST_REGISTRY{TEST_PROFILES, std::size(TEST_PROFILES)};
constexpr AolSafetyProfile VEHICLE_PROFILES[] = {HONDA_AOL_PROFILE, HONDA_STOCK_AOL_PROFILE, HONDA_NIDEC_AOL_PROFILE, HYUNDAI_AOL_PROFILE, HYUNDAI_CLASSIC_AOL_PROFILE, HYUNDAI_LEGACY_AOL_PROFILE, GM_AOL_PROFILE, FORD_AOL_PROFILE, MAZDA_AOL_PROFILE, CHRYSLER_AOL_PROFILE, TESLA_PREAP_AOL_PROFILE, TESLA_SCREEN_AOL_PROFILE, TOYOTA_AOL_PROFILE, NISSAN_AOL_PROFILE, VOLKSWAGEN_STOCK_AOL_PROFILE, VOLKSWAGEN_LONG_AOL_PROFILE, SUBARU_AOL_PROFILE};
constexpr AolSafetyProfile VEHICLE_PROFILES[] = {HONDA_AOL_PROFILE, HONDA_STOCK_AOL_PROFILE, HONDA_NIDEC_AOL_PROFILE, HYUNDAI_AOL_PROFILE, HYUNDAI_CLASSIC_AOL_PROFILE, HYUNDAI_LEGACY_AOL_PROFILE, GM_AOL_PROFILE, FORD_AOL_PROFILE, MAZDA_AOL_PROFILE, CHRYSLER_AOL_PROFILE, TESLA_PREAP_AOL_PROFILE, TESLA_SCREEN_AOL_PROFILE, TOYOTA_AOL_PROFILE, NISSAN_AOL_PROFILE, VOLKSWAGEN_STOCK_AOL_PROFILE, VOLKSWAGEN_LONG_AOL_PROFILE, VOLKSWAGEN_MEB_AOL_PROFILE, SUBARU_AOL_PROFILE};
const AolProfileRegistry VEHICLE_REGISTRY{VEHICLE_PROFILES, std::size(VEHICLE_PROFILES)};
aol_safety_health_t status(uint8_t request = 0U, uint8_t permission = 0U) {
@@ -101,6 +101,18 @@ int main() {
assert(!aol_capable(subaru, 11U, VEHICLE_REGISTRY));
}
for (uint32_t word = 0U; word <= 65535U; ++word) {
const bool expected = word == 8U || word == 10U || word == 12U || word == 14U ||
word == 24U || word == 26U || word == 28U || word == 30U;
auto meb = status();
meb.safety_mode = 34U;
meb.safety_param = static_cast<uint16_t>(word);
assert(volkswagen_meb_aol_param(static_cast<uint16_t>(word)) == expected);
assert(aol_capable(meb, 34U, VEHICLE_REGISTRY) == expected);
meb.capability_flags = 0U;
assert(!aol_capable(meb, 34U, VEHICLE_REGISTRY));
}
for (uint32_t word = 0U; word <= 65535U; ++word) {
const bool stock = (word == 8U) || (word == 24U) || (word == 40U);
const bool longitudinal = (word == 9U) || (word == 25U) || (word == 41U);
@@ -380,42 +380,27 @@
"official_baseline_bytes": 368030,
"official_baseline_sha256": "a987635c68085a9c9c25dbb92e23dc859d6a1644cd84f4b968a047b95a1c9033",
"car_fingerprint": "TOYOTA_COROLLA_TSS2",
"bytes": 368039,
"sha256": "7025bbc80ccfc8be3b48d73cce929a19a68415cb90f208511a3ec188e5cfdbcd",
"bytes": 341545,
"sha256": "0ba8b7d4a23fa2f22e7354dc5dda6e07bbfdd2ae21c0a4f205e34e5b4fe567ae",
"input_sha256": "876100021b13ddd1a39efad5fdb4e7376ad5b298f64bf390b5f58742ffc6e429",
"input_bytes": 9272538,
"config_sha256": "560c1060f4d2e43810e7bbd55d5f361a828bdb027d60f308b2e1d614222a2078",
"contract": "Toyota lanes follow applied lateral activity; only LKAS_HUD 0x412 bus0 length8 byte0 mask0x14 lanes1 to0 differs from the prior reviewed comparison. Prior platform and safety contract remains.",
"producing_source_snapshot": "root-actual-4ca263696691268a5ed229a6e65f743e435b9d5d-ec59971db14d-toyota-hud-replay",
"contract": "Toyota EPS lateral availability extends to zero speed: steerAtStandstill equals minSteerSpeed<=0. Exactly two CarParams publications change false totrue; all other fields/messages and prior HUD/safety/tune contracts are retained.",
"producing_source_snapshot": "CI-1fdc8b89e8016be62df8e289d1f5f67561fdd6d7-CP-only-two-event-substitution",
"review_evidence": {
"ci_run": 38207706213,
"ci_job": 114674772155,
"head": "4ca263696691268a5ed229a6e65f743e435b9d5d",
"comparison_reference_sha256": "c6bbd05f9b77701618101cc689c3eee6c0c9e8642eb0fd4ef0f7ed8a442df32e",
"difference_count": 20,
"difference_fields": {
"sendcan.1.dat": 7,
"sendcan.2.dat": 9,
"sendcan.3.dat": 4
},
"actual_paired_proof": {
"path": "/audit/oct12-toyota-replay-root-output-v1/focused/profile/regen1455E3B4BDF_2025-04-09--03-26-06--0/actual-paired-proof.json",
"sha256": "6de5981bf3103f857a5f35a1a76fb18115daffaf6e08eea453912b0a33e17966",
"mode": 420,
"kind": "file"
},
"source_sha256": {
"opendbc_repo/opendbc/car/toyota/carcontroller.py": "a9c15c2fb31cf713216525ac6bc30648e489cfb2b853ef41dd678db8e8d7d6f7",
"opendbc_repo/opendbc/car/toyota/toyotacan.py": "64ba30a26411341b0b5ca8c62c6ae413e9a6713a1b24cc1ae711d3cbd31bb2fc",
"opendbc_repo/opendbc/car/toyota/tests/test_toyota.py": "e6a3ef017a3a6582e8ab771ab1db33f4f0edcc091ebad5f081c1ce8e0f4a024a",
"opendbc_repo/opendbc/dbc/generator/toyota/_toyota_2017.dbc": "818872b1c1034983f61a51460a9b175b4bfe01f3d34f537c3d21f420991a8a48",
"tools/ci/replay_fixture_cache.py": "f165d1d228f5e4600760ea41996b923c095964550b3ced6c3135a59ec7625821",
"tools/ci/replay-fixture-cache.json": "99fd6ddc839ae2f779c4e3ef0f529359cd6e19fc20f953da1c591b3af8456e15",
"tools/ci/replay_references.py": "75267bb70fc6391bb3a1bc2c9f36af2807cf838b2ffcd72c522ae861fd431832",
"openpilot/selfdrive/test/process_replay/test_processes.py": "559f228029d430f1c6d30ea0ccc9c60ebb98cf48df0fe1a7f80ef5a7c23394d1",
"openpilot/selfdrive/test/process_replay/compare_logs.py": "fb923cff883974dcdf2d701a0b8484c9d3c4d2192b018ef2d17aeac39dd24202",
"openpilot/selfdrive/test/process_replay/process_replay.py": "a6f18da56ba8521328d5a2ac57de5426ef2a3bf62f9e7887ff76ba1c52491c09"
}
"ci_run": 39895639195,
"ci_artifact": 12476424903,
"head": "1fdc8b89e8016be62df8e289d1f5f67561fdd6d7",
"causal_commit": "4e27abda7d",
"causal_assignment": "ret.steerAtStandstill = ret.minSteerSpeed <= 0.",
"full_CP_other_fields_exact": true,
"non_CP_message_payloads_byte_exact": true,
"only_changed_field": "carParams.steerAtStandstill",
"difference_count": 2,
"actual_CI_sha256": "79baa812dc2681c021965eaceedf721e7fcb85e315c1b855697ec62b9800656f",
"previous_reference_sha256": "7025bbc80ccfc8be3b48d73cce929a19a68415cb90f208511a3ec188e5cfdbcd",
"proof_sha256": "2ba6c13414ba96e6b64d96398022d232650e9e58c6a739cab39dc8952c067acf",
"no_new_ignores_or_tolerances": true
},
"previous_reviewed_expectation": {
"segment": "regen1455E3B4BDF|2025-04-09--03-26-06--0",
@@ -427,28 +412,76 @@
"official_baseline_bytes": 368030,
"official_baseline_sha256": "a987635c68085a9c9c25dbb92e23dc859d6a1644cd84f4b968a047b95a1c9033",
"car_fingerprint": "TOYOTA_COROLLA_TSS2",
"bytes": 368011,
"sha256": "c6bbd05f9b77701618101cc689c3eee6c0c9e8642eb0fd4ef0f7ed8a442df32e",
"bytes": 368039,
"sha256": "7025bbc80ccfc8be3b48d73cce929a19a68415cb90f208511a3ec188e5cfdbcd",
"input_sha256": "876100021b13ddd1a39efad5fdb4e7376ad5b298f64bf390b5f58742ffc6e429",
"input_bytes": 9272538,
"config_sha256": "560c1060f4d2e43810e7bbd55d5f361a828bdb027d60f308b2e1d614222a2078",
"contract": "Finalized Corolla native full-axis profile uses word 73 and alternative experience 32; every other compared Card output field remains unchanged.",
"producing_source_snapshot": "3dfea-production",
"contract": "Toyota lanes follow applied lateral activity; only LKAS_HUD 0x412 bus0 length8 byte0 mask0x14 lanes1 to0 differs from the prior reviewed comparison. Prior platform and safety contract remains.",
"producing_source_snapshot": "root-actual-4ca263696691268a5ed229a6e65f743e435b9d5d-ec59971db14d-toyota-hud-replay",
"review_evidence": {
"ci_run": 37730163667,
"ci_job": 113157245274,
"comparison_reference_sha256": "a987635c68085a9c9c25dbb92e23dc859d6a1644cd84f4b968a047b95a1c9033",
"difference_count": 2,
"ci_run": 38207706213,
"ci_job": 114674772155,
"head": "4ca263696691268a5ed229a6e65f743e435b9d5d",
"comparison_reference_sha256": "c6bbd05f9b77701618101cc689c3eee6c0c9e8642eb0fd4ef0f7ed8a442df32e",
"difference_count": 20,
"difference_fields": {
"carParams.alternativeExperience": 2
"sendcan.1.dat": 7,
"sendcan.2.dat": 9,
"sendcan.3.dat": 4
},
"difference_sha256": "46a4645f5d6fda6b02a81909ee5aa24cf427c4753ae7e8db7495985a1592e473",
"behavior_proof": {
"factory_safety_param": 73,
"factory_alternative_experience": 32
"actual_paired_proof": {
"path": "/audit/oct12-toyota-replay-root-output-v1/focused/profile/regen1455E3B4BDF_2025-04-09--03-26-06--0/actual-paired-proof.json",
"sha256": "6de5981bf3103f857a5f35a1a76fb18115daffaf6e08eea453912b0a33e17966",
"mode": 420,
"kind": "file"
},
"regression_tests": {
"openpilot/starpilot/aol/tests/test_toyota.py": "6cb1fddc0a6f5f603f909481dca2c97260bcb60fbbee388e461cc58f4cf053fe"
"source_sha256": {
"opendbc_repo/opendbc/car/toyota/carcontroller.py": "a9c15c2fb31cf713216525ac6bc30648e489cfb2b853ef41dd678db8e8d7d6f7",
"opendbc_repo/opendbc/car/toyota/toyotacan.py": "64ba30a26411341b0b5ca8c62c6ae413e9a6713a1b24cc1ae711d3cbd31bb2fc",
"opendbc_repo/opendbc/car/toyota/tests/test_toyota.py": "e6a3ef017a3a6582e8ab771ab1db33f4f0edcc091ebad5f081c1ce8e0f4a024a",
"opendbc_repo/opendbc/dbc/generator/toyota/_toyota_2017.dbc": "818872b1c1034983f61a51460a9b175b4bfe01f3d34f537c3d21f420991a8a48",
"tools/ci/replay_fixture_cache.py": "f165d1d228f5e4600760ea41996b923c095964550b3ced6c3135a59ec7625821",
"tools/ci/replay-fixture-cache.json": "99fd6ddc839ae2f779c4e3ef0f529359cd6e19fc20f953da1c591b3af8456e15",
"tools/ci/replay_references.py": "75267bb70fc6391bb3a1bc2c9f36af2807cf838b2ffcd72c522ae861fd431832",
"openpilot/selfdrive/test/process_replay/test_processes.py": "559f228029d430f1c6d30ea0ccc9c60ebb98cf48df0fe1a7f80ef5a7c23394d1",
"openpilot/selfdrive/test/process_replay/compare_logs.py": "fb923cff883974dcdf2d701a0b8484c9d3c4d2192b018ef2d17aeac39dd24202",
"openpilot/selfdrive/test/process_replay/process_replay.py": "a6f18da56ba8521328d5a2ac57de5426ef2a3bf62f9e7887ff76ba1c52491c09"
}
},
"previous_reviewed_expectation": {
"segment": "regen1455E3B4BDF|2025-04-09--03-26-06--0",
"process": "card",
"file": "regen1455E3B4BDF_2025-04-09--03-26-06--0_card.zst",
"official_baseline_commit": "ec95db3f1fa19f76940497fedfdb62e09ea19912",
"official_artifact_commit": "9ae451c262c7402c9623f3baa1580c6242d6777b",
"official_baseline_url": "https://raw.githubusercontent.com/commaai/ci-artifacts/9ae451c262c7402c9623f3baa1580c6242d6777b/regen1455E3B4BDF_2025-04-09--03-26-06--0_card_ec95db3f1fa19f76940497fedfdb62e09ea19912.zst",
"official_baseline_bytes": 368030,
"official_baseline_sha256": "a987635c68085a9c9c25dbb92e23dc859d6a1644cd84f4b968a047b95a1c9033",
"car_fingerprint": "TOYOTA_COROLLA_TSS2",
"bytes": 368011,
"sha256": "c6bbd05f9b77701618101cc689c3eee6c0c9e8642eb0fd4ef0f7ed8a442df32e",
"input_sha256": "876100021b13ddd1a39efad5fdb4e7376ad5b298f64bf390b5f58742ffc6e429",
"input_bytes": 9272538,
"config_sha256": "560c1060f4d2e43810e7bbd55d5f361a828bdb027d60f308b2e1d614222a2078",
"contract": "Finalized Corolla native full-axis profile uses word 73 and alternative experience 32; every other compared Card output field remains unchanged.",
"producing_source_snapshot": "3dfea-production",
"review_evidence": {
"ci_run": 37730163667,
"ci_job": 113157245274,
"comparison_reference_sha256": "a987635c68085a9c9c25dbb92e23dc859d6a1644cd84f4b968a047b95a1c9033",
"difference_count": 2,
"difference_fields": {
"carParams.alternativeExperience": 2
},
"difference_sha256": "46a4645f5d6fda6b02a81909ee5aa24cf427c4753ae7e8db7495985a1592e473",
"behavior_proof": {
"factory_safety_param": 73,
"factory_alternative_experience": 32
},
"regression_tests": {
"openpilot/starpilot/aol/tests/test_toyota.py": "6cb1fddc0a6f5f603f909481dca2c97260bcb60fbbee388e461cc58f4cf053fe"
}
}
}
}
@@ -466,36 +499,69 @@
"input_bytes": 10846201,
"input_sha256": "6c028c0d40328fc00fa12acf91dba30d9f152e89d99a7089aadac62f6c7d5d64",
"config_sha256": "560c1060f4d2e43810e7bbd55d5f361a828bdb027d60f308b2e1d614222a2078",
"bytes": 305683,
"sha256": "ca61046afe52ed7707ccb39752471d6ae0e377151e4075c9a39c01e07b18c6bf",
"contract": "Toyota lanes follow applied lateral activity; only LKAS_HUD 0x412 bus0 length8 byte0 mask0x14 lanes1 to0 differs from the prior reviewed comparison. Prior platform and safety contract remains.",
"producing_source_snapshot": "root-actual-4ca263696691268a5ed229a6e65f743e435b9d5d-ec59971db14d-toyota-hud-replay",
"bytes": 285508,
"sha256": "f36d45ae9966c8705b529137a0cc183e35de68269f65ab2fab4a2c1b3e3ed95b",
"contract": "Toyota EPS lateral availability extends to zero speed: steerAtStandstill equals minSteerSpeed<=0. Exactly two CarParams publications change false totrue; all other fields/messages and prior HUD/safety/tune contracts are retained.",
"producing_source_snapshot": "CI-1fdc8b89e8016be62df8e289d1f5f67561fdd6d7-CP-only-two-event-substitution",
"review_evidence": {
"ci_run": 38207706213,
"ci_job": 114674772155,
"head": "4ca263696691268a5ed229a6e65f743e435b9d5d",
"comparison_reference_sha256": "9688b1b72e3c992cb6c72e41e1607d2a38625e141caed9b356fadea5ee53aea2",
"difference_count": 222,
"difference_fields": {
"sendcan.1.dat": 222
},
"actual_paired_proof": {
"path": "/audit/oct12-toyota-replay-root-output-v1/focused/profile/regen218A4DCFAA1_2025-04-08--22-57-51--0/actual-paired-proof.json",
"sha256": "2530635df4b9014ce3fd4e70fe59b8d6655196f4fc2eddc5736fb87721d8c381",
"mode": 420,
"kind": "file"
},
"source_sha256": {
"opendbc_repo/opendbc/car/toyota/carcontroller.py": "a9c15c2fb31cf713216525ac6bc30648e489cfb2b853ef41dd678db8e8d7d6f7",
"opendbc_repo/opendbc/car/toyota/toyotacan.py": "64ba30a26411341b0b5ca8c62c6ae413e9a6713a1b24cc1ae711d3cbd31bb2fc",
"opendbc_repo/opendbc/car/toyota/tests/test_toyota.py": "e6a3ef017a3a6582e8ab771ab1db33f4f0edcc091ebad5f081c1ce8e0f4a024a",
"opendbc_repo/opendbc/dbc/generator/toyota/_toyota_2017.dbc": "818872b1c1034983f61a51460a9b175b4bfe01f3d34f537c3d21f420991a8a48",
"tools/ci/replay_fixture_cache.py": "f165d1d228f5e4600760ea41996b923c095964550b3ced6c3135a59ec7625821",
"tools/ci/replay-fixture-cache.json": "99fd6ddc839ae2f779c4e3ef0f529359cd6e19fc20f953da1c591b3af8456e15",
"tools/ci/replay_references.py": "75267bb70fc6391bb3a1bc2c9f36af2807cf838b2ffcd72c522ae861fd431832",
"openpilot/selfdrive/test/process_replay/test_processes.py": "559f228029d430f1c6d30ea0ccc9c60ebb98cf48df0fe1a7f80ef5a7c23394d1",
"openpilot/selfdrive/test/process_replay/compare_logs.py": "fb923cff883974dcdf2d701a0b8484c9d3c4d2192b018ef2d17aeac39dd24202",
"openpilot/selfdrive/test/process_replay/process_replay.py": "a6f18da56ba8521328d5a2ac57de5426ef2a3bf62f9e7887ff76ba1c52491c09"
"ci_run": 39895639195,
"ci_artifact": 12476424903,
"head": "1fdc8b89e8016be62df8e289d1f5f67561fdd6d7",
"causal_commit": "4e27abda7d",
"causal_assignment": "ret.steerAtStandstill = ret.minSteerSpeed <= 0.",
"full_CP_other_fields_exact": true,
"non_CP_message_payloads_byte_exact": true,
"only_changed_field": "carParams.steerAtStandstill",
"difference_count": 2,
"actual_CI_sha256": "77c1ae09380fd6ae657b96dc89b1391fa1552b98de1adc4a284bdea2dee38bc4",
"previous_reference_sha256": "ca61046afe52ed7707ccb39752471d6ae0e377151e4075c9a39c01e07b18c6bf",
"proof_sha256": "2ba6c13414ba96e6b64d96398022d232650e9e58c6a739cab39dc8952c067acf",
"no_new_ignores_or_tolerances": true
},
"previous_reviewed_expectation": {
"segment": "regen218A4DCFAA1|2025-04-08--22-57-51--0",
"process": "card",
"file": "regen218A4DCFAA1_2025-04-08--22-57-51--0_card.zst",
"official_baseline_commit": "ec95db3f1fa19f76940497fedfdb62e09ea19912",
"official_artifact_commit": "9ae451c262c7402c9623f3baa1580c6242d6777b",
"official_baseline_url": "https://raw.githubusercontent.com/commaai/ci-artifacts/9ae451c262c7402c9623f3baa1580c6242d6777b/regen218A4DCFAA1_2025-04-08--22-57-51--0_card_ec95db3f1fa19f76940497fedfdb62e09ea19912.zst",
"official_baseline_bytes": 305708,
"official_baseline_sha256": "9688b1b72e3c992cb6c72e41e1607d2a38625e141caed9b356fadea5ee53aea2",
"car_fingerprint": "TOYOTA_PRIUS",
"input_bytes": 10846201,
"input_sha256": "6c028c0d40328fc00fa12acf91dba30d9f152e89d99a7089aadac62f6c7d5d64",
"config_sha256": "560c1060f4d2e43810e7bbd55d5f361a828bdb027d60f308b2e1d614222a2078",
"bytes": 305683,
"sha256": "ca61046afe52ed7707ccb39752471d6ae0e377151e4075c9a39c01e07b18c6bf",
"contract": "Toyota lanes follow applied lateral activity; only LKAS_HUD 0x412 bus0 length8 byte0 mask0x14 lanes1 to0 differs from the prior reviewed comparison. Prior platform and safety contract remains.",
"producing_source_snapshot": "root-actual-4ca263696691268a5ed229a6e65f743e435b9d5d-ec59971db14d-toyota-hud-replay",
"review_evidence": {
"ci_run": 38207706213,
"ci_job": 114674772155,
"head": "4ca263696691268a5ed229a6e65f743e435b9d5d",
"comparison_reference_sha256": "9688b1b72e3c992cb6c72e41e1607d2a38625e141caed9b356fadea5ee53aea2",
"difference_count": 222,
"difference_fields": {
"sendcan.1.dat": 222
},
"actual_paired_proof": {
"path": "/audit/oct12-toyota-replay-root-output-v1/focused/profile/regen218A4DCFAA1_2025-04-08--22-57-51--0/actual-paired-proof.json",
"sha256": "2530635df4b9014ce3fd4e70fe59b8d6655196f4fc2eddc5736fb87721d8c381",
"mode": 420,
"kind": "file"
},
"source_sha256": {
"opendbc_repo/opendbc/car/toyota/carcontroller.py": "a9c15c2fb31cf713216525ac6bc30648e489cfb2b853ef41dd678db8e8d7d6f7",
"opendbc_repo/opendbc/car/toyota/toyotacan.py": "64ba30a26411341b0b5ca8c62c6ae413e9a6713a1b24cc1ae711d3cbd31bb2fc",
"opendbc_repo/opendbc/car/toyota/tests/test_toyota.py": "e6a3ef017a3a6582e8ab771ab1db33f4f0edcc091ebad5f081c1ce8e0f4a024a",
"opendbc_repo/opendbc/dbc/generator/toyota/_toyota_2017.dbc": "818872b1c1034983f61a51460a9b175b4bfe01f3d34f537c3d21f420991a8a48",
"tools/ci/replay_fixture_cache.py": "f165d1d228f5e4600760ea41996b923c095964550b3ced6c3135a59ec7625821",
"tools/ci/replay-fixture-cache.json": "99fd6ddc839ae2f779c4e3ef0f529359cd6e19fc20f953da1c591b3af8456e15",
"tools/ci/replay_references.py": "75267bb70fc6391bb3a1bc2c9f36af2807cf838b2ffcd72c522ae861fd431832",
"openpilot/selfdrive/test/process_replay/test_processes.py": "559f228029d430f1c6d30ea0ccc9c60ebb98cf48df0fe1a7f80ef5a7c23394d1",
"openpilot/selfdrive/test/process_replay/compare_logs.py": "fb923cff883974dcdf2d701a0b8484c9d3c4d2192b018ef2d17aeac39dd24202",
"openpilot/selfdrive/test/process_replay/process_replay.py": "a6f18da56ba8521328d5a2ac57de5426ef2a3bf62f9e7887ff76ba1c52491c09"
}
}
}
},
@@ -2,6 +2,8 @@ from collections.abc import Callable
from openpilot.common.params import Params
from openpilot.starpilot.car.hyundai.aol import ioniq6_settings_capable
from opendbc.car.volkswagen.meb_modes import MEB_CARS, settings_capable as meb_settings_capable
from openpilot.starpilot.car.volkswagen.preferences import stock_long_requested
from openpilot.selfdrive.ui.widgets.ssh_key import ssh_key_item
from openpilot.selfdrive.ui.ui_state import ui_state
from openpilot.system.ui.widgets import Widget
@@ -140,8 +142,9 @@ class DeveloperLayout(Widget):
# CP gating
if ui_state.CP is not None:
alpha_avail = ui_state.CP.alphaLongitudinalAvailable or ioniq6_settings_capable(ui_state.CP)
if not alpha_avail or self._is_release:
alpha_avail = (meb_settings_capable(ui_state.CP) if ui_state.CP.brand == 'volkswagen' and ui_state.CP.carFingerprint in MEB_CARS
else ui_state.CP.alphaLongitudinalAvailable or ioniq6_settings_capable(ui_state.CP))
if not alpha_avail or (self._is_release and not meb_settings_capable(ui_state.CP)):
self._alpha_long_toggle.set_visible(False)
else:
self._alpha_long_toggle.set_visible(True)
@@ -166,7 +169,9 @@ class DeveloperLayout(Widget):
("ShowDebugInfo", self._ui_debug_toggle),
("CellularDisabled", self._cellular_disabled_toggle),
):
item.action_item.set_state(self._params.get_bool(key))
value = (not stock_long_requested(self._params) if key == "AlphaLongitudinalEnabled" and meb_settings_capable(ui_state.CP)
else self._params.get_bool(key))
item.action_item.set_state(value)
def _on_enable_ui_debug(self, state: bool):
self._params.put_bool("ShowDebugInfo", state, block=True)
@@ -209,7 +214,8 @@ class DeveloperLayout(Widget):
def confirm_callback(result: DialogResult):
if result == DialogResult.CONFIRM:
if (not ui_state.is_offroad() or ui_state.CP is None or
not (ui_state.CP.alphaLongitudinalAvailable or ioniq6_settings_capable(ui_state.CP))):
not (meb_settings_capable(ui_state.CP) if ui_state.CP.brand == 'volkswagen' and ui_state.CP.carFingerprint in MEB_CARS
else ui_state.CP.alphaLongitudinalAvailable or ioniq6_settings_capable(ui_state.CP))):
self._alpha_long_toggle.action_item.set_state(False)
return
self._params.put_bool("AlphaLongitudinalEnabled", True, block=True)
@@ -8,11 +8,13 @@ from openpilot.system.ui.lib.application import gui_app
from openpilot.selfdrive.ui.layouts.settings.common import restart_needed_callback
from openpilot.selfdrive.ui.ui_state import ui_state
from openpilot.starpilot.car.hyundai.aol import ioniq6_settings_capable
from opendbc.car.volkswagen.meb_modes import MEB_CARS, settings_capable as meb_settings_capable
from openpilot.starpilot.car.volkswagen.preferences import stock_long_requested
from openpilot.selfdrive.ui.widgets.ssh_key import SshKeyFetcher
class AlphaLongConfirmPage(NavScroller):
def __init__(self, on_confirm: Callable[[], None]):
def __init__(self, on_confirm: Callable[[], None], *, system_long_default: bool = False):
super().__init__()
accept = BigConfirmationCircleButton("enable alpha\nlongitudinal",
@@ -23,7 +25,8 @@ class AlphaLongConfirmPage(NavScroller):
GreyBigButton("enabling alpha longitudinal", "scroll to continue",
gui_app.texture("icons_mici/setup/warning.png", 64, 64)),
GreyBigButton("", "WARNING: alpha longitudinal control may disable Automatic Emergency Braking (AEB)"),
GreyBigButton("", "On this car, openpilot defaults to the stock system's built-in ACC."),
GreyBigButton("", "Use StarPilot speed control instead of stock adaptive cruise control." if system_long_default else
"On this car, openpilot defaults to the stock system's built-in ACC."),
GreyBigButton("", "This requests openpilot longitudinal control after a verified vehicle takeover at startup."),
accept,
])
@@ -166,8 +169,9 @@ class DeveloperLayoutMici(NavScroller):
# CP gating
if ui_state.CP is not None:
alpha_avail = ui_state.CP.alphaLongitudinalAvailable or ioniq6_settings_capable(ui_state.CP)
if not alpha_avail or ui_state.is_release:
alpha_avail = (meb_settings_capable(ui_state.CP) if ui_state.CP.brand == 'volkswagen' and ui_state.CP.carFingerprint in MEB_CARS
else ui_state.CP.alphaLongitudinalAvailable or ioniq6_settings_capable(ui_state.CP))
if not alpha_avail or (ui_state.is_release and not meb_settings_capable(ui_state.CP)):
self._alpha_long_toggle.set_visible(False)
else:
self._alpha_long_toggle.set_visible(True)
@@ -177,7 +181,9 @@ class DeveloperLayoutMici(NavScroller):
# Refresh toggles from params to mirror external changes
for key, item in self._refresh_toggles:
item.set_checked(ui_state.params.get_bool(key))
value = (not stock_long_requested(ui_state.params) if key == "AlphaLongitudinalEnabled" and meb_settings_capable(ui_state.CP)
else ui_state.params.get_bool(key))
item.set_checked(value)
def _on_joystick_debug_mode(self, state: bool):
ui_state.params.put_bool("JoystickDebugMode", state, block=True)
@@ -187,7 +193,8 @@ class DeveloperLayoutMici(NavScroller):
def _on_alpha_long_enabled(self, state: bool):
def do_toggle(_state: bool):
if (not ui_state.is_offroad() or (_state and (ui_state.CP is None or
not (ui_state.CP.alphaLongitudinalAvailable or ioniq6_settings_capable(ui_state.CP))))):
not (meb_settings_capable(ui_state.CP) if ui_state.CP.brand == 'volkswagen' and ui_state.CP.carFingerprint in MEB_CARS
else ui_state.CP.alphaLongitudinalAvailable or ioniq6_settings_capable(ui_state.CP))))):
self._update_toggles()
return
ui_state.params.put_bool("AlphaLongitudinalEnabled", _state, block=True)
@@ -197,6 +204,8 @@ class DeveloperLayoutMici(NavScroller):
if state:
# Don't show enabled state until confirm
self._alpha_long_toggle.set_checked(False)
gui_app.push_widget(AlphaLongConfirmPage(lambda: do_toggle(True)))
page = (AlphaLongConfirmPage(lambda: do_toggle(True), system_long_default=True) if meb_settings_capable(ui_state.CP)
else AlphaLongConfirmPage(lambda: do_toggle(True)))
gui_app.push_widget(page)
else:
do_toggle(False)
@@ -0,0 +1,288 @@
"""Actual SD.step/Controls.publish/Card.controls_update + native TX. Root runtime only.
Synthetic physical CAN/transport; no native/axis/CarParams permission is fabricated.
The shared actual-caller harness serializes measured native booleans/inventory and
retains the real family events/state machine. This is not physical road proof.
"""
from contextlib import ExitStack
import shutil
import tempfile
from pathlib import Path
from unittest import mock
import pytest
from opendbc.can import CANPacker, CANParser
from opendbc.can.dbc import DBC as ParsedDBC
from opendbc.car import Bus, structs
from opendbc.car.volkswagen.interface import CarInterface
from opendbc.car.volkswagen.meb_modes import profile_word
from opendbc.car.volkswagen.values import DBC
from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences
from opendbc.safety.tests.common import make_msg
from opendbc.safety.tests.libsafety import libsafety_py
from openpilot.cereal import messaging
from openpilot.common.params import Params
from openpilot.common.prefix import OpenpilotPrefix
from openpilot.selfdrive.car import card as card_module
from openpilot.starpilot.aol.tests.stock_lateral_test_helpers import authority_case
@pytest.mark.parametrize('identity', ('VOLKSWAGEN_ID4_MK1', 'VOLKSWAGEN_ID4_MK2'))
@pytest.mark.parametrize('stock', (False, True))
@pytest.mark.parametrize('alt_gear', (False, True))
@pytest.mark.parametrize('offset,fixed', ((0, False), (4, True)))
def test_meb_full_engagement_brake_reverse_drive_actual_callers(identity, stock, alt_gear, offset, fixed, *, fault_phase=False):
with ExitStack() as cleanup:
cleanup.enter_context(OpenpilotPrefix())
cleanup.enter_context(mock.patch.dict('os.environ', {'SIMULATION': '0', 'AOL_REPLAY_RUNTIME': '1'}))
params = Params()
params.put_bool('AlwaysOnLateral', True, block=True)
params.put('MainCruiseButtonControl', 0, block=True)
params.put('AolBrakePauseSpeedMps', 5., block=True)
params.put_bool('DisableOpenpilotLongitudinal', stock, block=True)
fp = {bus: {} for bus in range(8)}
fp[offset + 1][0x13D] = 32
fp[offset][0x25D] = 8 # Actual KLR source selects the existing stock Emergency Assist contract.
if alt_gear:
fp[offset][0x3DC] = 8
cp = CarInterface.get_params(identity, fp, [], False, False, False)
VehicleStartupPreferences.read(params, enabled=True).prepare(cp, fingerprints=fp)
cp.fingerprintSource = structs.CarParams.FingerprintSource.fixed if fixed else structs.CarParams.FingerprintSource.fw
# Root preloads the matching compiled library. This runner never invokes a compiler.
assert 'libsafety' in libsafety_py.__dict__, 'Root must preload the source-matched native library'
native = libsafety_py.libsafety
quiet = None
if offset:
temporary = tempfile.TemporaryDirectory(prefix='vw-native-pair-')
cleanup.callback(temporary.cleanup)
copy = Path(temporary.name) / 'quiet.so'
shutil.copy2(libsafety_py.libpath, copy)
quiet = libsafety_py.ffi.dlopen(str(copy))
cleanup.callback(libsafety_py.ffi.dlclose, quiet)
quiet.init_tests()
quiet.set_alternative_experience(32)
quiet.set_safety_hooks(structs.CarParams.SafetyModel.noOutput, 0)
assert not quiet.get_controls_allowed()
# The caller harness receives genuine current noOutput state for the leading slot.
from openpilot.starpilot.aol.wire import PandaSlot
def measured_slot(index, serial, mode, word, experience, controls_allowed, invalid, lost, faults):
if index == 0:
assert mode == quiet.get_current_safety_mode() and word == quiet.get_current_safety_param()
controls_allowed = bool(quiet.get_controls_allowed())
return PandaSlot(index, serial, mode, word, experience, controls_allowed, invalid, lost, faults)
cleanup.enter_context(mock.patch('openpilot.starpilot.aol.tests.test_ioniq_availability.PandaSlot', side_effect=measured_slot))
case = authority_case()
cleanup.callback(case.doCleanups)
# The harness marks its SD CP. Card must independently finalize the original normal factory CP.
card_cp = structs.CarParams(**cp.to_dict())
sd, controls, step = case._authority_callers(False, cp=cp, genuine_events=True)
card = card_module.Car(CI=CarInterface(card_cp), RI=CarInterface.RadarInterface(card_cp))
assert card.CP.to_dict() == sd.CP.to_dict()
if offset:
original_update = sd.sm.update_msgs
def measured_pandas(timestamp, messages):
actual = []
for envelope in messages:
if envelope.which() == 'pandaStates':
envelope = envelope.as_builder()
envelope.pandaStates[0].controlsAllowed = bool(quiet.get_controls_allowed())
actual.append(envelope)
return original_update(timestamp, actual)
cleanup.enter_context(mock.patch.object(sd.sm, 'update_msgs', side_effect=measured_pandas))
# Real DM messages, not a direct edit of CarControl escalation or native permissions.
def dm_samples(update):
next_due = {}
def receive(timestamp, messages):
messages = list(messages)
if timestamp >= 10. and not any(envelope.which() == 'driverMonitoringState' for envelope in messages):
messages.append(messaging.new_message('driverMonitoringState', valid=True, logMonoTime=int(timestamp * 1e9)).as_reader())
actual = []
for envelope in messages:
service = envelope.which()
if service in ('driverMonitoringState', 'aolSafetyWire'):
stamp = int(timestamp * 1e9)
if stamp < next_due.get(service, 0):
continue
next_due[service] = stamp + (50_000_000 if service == 'driverMonitoringState' else 100_000_000)
if envelope.which() == 'driverMonitoringState':
envelope = envelope.as_builder()
envelope.driverMonitoringState.noResponseForceDecel = int(envelope.logMonoTime) >= (12_200_000_000 if fault_phase else 11_100_000_000)
actual.append(envelope)
return update(timestamp, actual)
return receive
cleanup.enter_context(mock.patch.object(sd.sm, 'update_msgs', side_effect=dm_samples(sd.sm.update_msgs)))
cleanup.enter_context(mock.patch.object(controls.sm, 'update_msgs', side_effect=dm_samples(controls.sm.update_msgs)))
# Explicit ambient producers added since the original shared caller fixture.
# Keep production SubMaster health/frequency checks; never ignore a required service.
from openpilot.cereal.services import SERVICE_LIST
ambient_services = ('alertDebug', 'lateralManeuverPlan', 'laneChangeAssistWire',
'starpilotLongitudinalPlan', 'slcState', 'gpsLocation')
original_sd_update = sd.sm.update_msgs
ambient_due = {}
def ambient_update(timestamp, messages):
stamp = int(timestamp * 1e9)
actual = list(messages)
present = {message.which() for message in actual}
for service in ambient_services:
if service in sd.sm.services and service not in present and stamp >= ambient_due.get(service, 0):
frequency = SERVICE_LIST[service].frequency
ambient_due[service] = stamp + (round(1e9 / min(frequency, 100)) if frequency > 0 else 0)
actual.append(messaging.new_message(service, 0 if service == 'laneChangeAssistWire' else None, valid=True, logMonoTime=stamp).as_reader())
return original_sd_update(timestamp, actual)
cleanup.enter_context(mock.patch.object(sd.sm, 'update_msgs', side_effect=ambient_update))
for prime in range(-110, 0):
ambient_update((10_000_000_000 + prime * 10_000_000) / 1e9, [])
card.slc_producer_session = 'authority-card'
class Publisher(messaging.PubMaster):
def __init__(self):
self.messages = {}
def send(self, s, dat):
if not isinstance(dat, bytes):
dat = dat.to_bytes()
self.messages[s] = messaging.log_from_bytes(dat)
card.pm = Publisher()
native = libsafety_py.libsafety
cleanup.callback(native.init_tests)
native.init_tests()
native.set_alternative_experience(32)
native.set_safety_hooks(34, profile_word(cp))
native.set_aol_test_heartbeat(True)
card.CI.update([])
dbc_name = DBC[identity][Bus.pt]
dbc = ParsedDBC(dbc_name)
packer = CANPacker(dbc_name)
emitted, lat_commands, long_commands = [], [], []
klr_escalated = []
fault_trace = []
request = 0
dm_tick = 222 if fault_phase else 112
for tick in range(240 if fault_phase else 120):
now = 10_000_000_000 + tick * 10_000_000
native.set_timer(1_000_000 + tick * 10_000)
if quiet is not None:
quiet.set_timer(1_000_000 + tick * 10_000)
quiet.safety_tick()
assert quiet.get_current_safety_mode() == int(structs.CarParams.SafetyModel.noOutput)
assert not quiet.get_controls_allowed()
assert native.get_current_safety_mode() != quiet.get_current_safety_mode()
brake = 65 <= tick < 75
gas = 45 <= tick < 50
reverse = 85 <= tick < 95
ordinary = 21 <= tick < 65 or (fault_phase and tick >= 200)
overrides = {
'Motor_51': {'TSK_Status': 3 if ordinary and stock else 2, 'Accel_Pedal_Pressure': 10 if gas else 0},
'Motor_54': {'Engine_On': 1},
'LH_EPS_03': {'EPS_Lenkmoment': 0},
'QFK_01': {'LatCon_HCA_Status': (7 if fault_phase and 120 <= tick < 125 else 3 if fault_phase and 160 <= tick < 162 else 2), 'Curvature': 0},
'Motor_14': {'MO_Fahrer_bremst': int(brake)},
'Getriebe_11': {'GE_Fahrstufe': 6 if reverse else 8},
'Gateway_73': {'GE_Fahrstufe': 6 if reverse else 8},
'ESC_51': {name + '_Radgeschw': 72 for name in ('VL', 'VR', 'HL', 'HR')},
'ESP_21': {'ESP_v_Signal': 72},
'Airbag_02': {'AB_Gurtschloss_FA': 0 if 105 <= tick < 111 else 3},
'GRA_ACC_01': {'GRA_Hauptschalter': 1, 'GRA_Tip_Setzen': int(tick == 20 or (fault_phase and tick == 200))},
'ACC_19': {'ACC_Gesetzte_Zeitluecke': 1, 'ACC_Wunschgeschw_02': 72},
}
incoming = []
for parser in card.CI.can_parsers.values():
for name in tuple(parser.vl):
if isinstance(name, str) and name in dbc.name_to_msg:
fields: dict[str, float] = dict(overrides.get(name, {}))
if 'COUNTER' in dbc.name_to_msg[name].sigs:
fields['COUNTER'] = tick % 16
incoming.append(packer.make_can_msg(name, parser.bus, fields))
for addr, data, bus in incoming:
native.safety_rx_hook(make_msg(bus - offset, addr, dat=data))
native.safety_tick()
native.aol_set_host_request(request)
permission = native.aol_get_permission_mask()
original_message = messaging.new_message
def clocked_card_message(*args, original_message=original_message, now=now, **kwargs):
message = original_message(*args, **kwargs)
message.logMonoTime = now
return message
with (mock.patch.object(card_module.messaging, 'new_message', side_effect=clocked_card_message),
mock.patch.object(card_module.messaging, 'drain_sock_raw', return_value=[b'fixture']),
mock.patch.object(card_module, 'can_capnp_to_list', return_value=[(now, incoming)]),
mock.patch.object(card_module.time, 'monotonic_ns', return_value=now),
mock.patch.object(card_module.time, 'clock_gettime_ns', return_value=now),
mock.patch.object(card_module, 'REPLAY', False), mock.patch.object(card.sm, 'update'),
mock.patch.object(card.RI, 'update', return_value=None)):
state, radar = card.state_update()
card.state_publish(state, radar)
if tick >= 15:
assert state.canValid and not state.canTimeout
intent = card.pm.messages['aolIntentWire'].as_builder()
axis, selfdrive, command = step(tick, observed_state=state, intent_message=intent,
gas=state.gasPressed, steering=state.steeringPressed, brake=state.brakePressed,
allowed=bool(permission & 1), long_allowed=bool(permission & 2),
native_request_lat=bool(request & 1), native_request_long=bool(request & 2),
panda_controls_allowed=native.get_controls_allowed())
if tick == 60 and not stock:
assert native.get_controls_allowed() and axis.desiredLongitudinal and command.longActive
if tick >= dm_tick:
assert command.driverMonitoringEscalation
request = int(axis.desiredLateral) | (int(axis.desiredLongitudinal) << 1)
# Feed the exact serialized Controls publication into the actual Card caller.
frames = [controls.pm.messages['carControl'], sd.pm.messages['aolAxisState'],
sd.pm.messages['onroadEvents'], sd.pm.messages['selfdriveState']]
for name in ('aolSafetyWire', 'pandaStates', 'deviceState'):
envelope = messaging.new_message(name, len(sd.sm[name]) if name in ('pandaStates', 'aolSafetyWire') else None,
valid=sd.sm.valid[name], logMonoTime=sd.sm.logMonoTime[name])
setattr(envelope, name, sd.sm[name])
frames.append(envelope.as_reader())
card.sm.update_msgs(now / 1e9, frames)
assert card.sm['carControl'].driverMonitoringEscalation == command.driverMonitoringEscalation
sent = []
with (mock.patch.object(card_module.time, 'monotonic_ns', return_value=now),
mock.patch.object(card_module.time, 'clock_gettime_ns', return_value=now),
mock.patch.object(card_module, 'REPLAY', False),
mock.patch.object(card, 'publish_sendcan', side_effect=lambda values, sent=sent, **kw: sent.extend(values))):
card.controls_update(state, card.sm['carControl'])
for addr, data, bus in sent:
assert offset <= bus < offset + 4
assert native.safety_tx_hook(make_msg(bus - offset, addr, dat=data)), (tick, hex(addr), data.hex(), bus)
if tick >= dm_tick and any(addr == 0x25D and bus == offset + 2 for addr, _, bus in sent):
# Decode the current output payload independently of this synthetic RX stream's counters.
klr_observer = CANParser(dbc_name, [('KLR_01', 0)], offset + 2)
klr_observer.update([(now, sent)])
klr_escalated.append(klr_observer.vl['KLR_01']['KLR_Touchauswertung'])
if quiet is not None:
for addr, data, bus in sent:
assert not quiet.safety_tx_hook(make_msg(bus - offset, addr, dat=data))
emitted.extend(sent)
if fault_phase and tick in (119, 120, 124, 125, 129, 130, 132, 139, 140, 145, 150, 159, 160, 169, 170, 175, 180):
fault_trace.append((tick, permission, card.aol_card_intent.allowed_latch, state.steerFaultTemporary,
state.steerFaultPermanent, axis.desiredLateral, axis.lateralActive,
axis.nativeAcknowledged, command.latActive, str(axis.faultReason),
card.aol_card_intent._fault_inhibit, card.aol_card_intent._native_rejection_ns))
lat_commands.append(command.latActive)
long_commands.append(command.longActive)
assert any(lat_commands[35:60])
assert any(lat_commands[76:84]) # braking released long, not qualified lateral intent
assert not any(lat_commands[86:95])
assert any(lat_commands[100:])
assert any(lat_commands[106:110]) # Existing qualified unbelted policy, with current native permission.
assert not any(long_commands[76:200 if fault_phase else None]) # Brake disables long; it never restores automatically.
if fault_phase:
assert not any(lat_commands[121:125])
if not any(lat_commands[145:155]):
print('MEB_FAULT_TRACE', fault_trace)
assert any(lat_commands[145:155]), fault_trace # Temporary EPS rejection retains intent.
assert not any(lat_commands[161:200]) # A recovered permanent fault cannot restore armed intent.
assert any(lat_commands[212:220]) # Fresh ordinary driver engagement rearms it.
if not stock:
assert any(long_commands[212:220])
assert emitted
assert klr_escalated and all(value == 0 for value in klr_escalated)
if not stock:
assert any(long_commands[35:60])
else:
assert not any(long_commands)
@pytest.mark.parametrize('identity', ('VOLKSWAGEN_ID4_MK1', 'VOLKSWAGEN_ID4_MK2'))
@pytest.mark.parametrize('stock', (False, True))
def test_meb_actual_eps_rejection_permanent_fault_and_fresh_ordinary_rearm(identity, stock):
test_meb_full_engagement_brake_reverse_drive_actual_callers(identity, stock, False, 0, False, fault_phase=True)
+16 -2
View File
@@ -450,6 +450,7 @@ class TransportPauseFeedback:
from openpilot.starpilot.car.nissan.aol import qualified as nissan_qualified
from opendbc.car.ford.aol import qualified as ford_qualified
from opendbc.car.rivian.stock_aol import qualified as rivian_qualified
from openpilot.starpilot.car.volkswagen.aol import qualified as volkswagen_qualified
from opendbc.car.mazda.stock_aol import qualified as mazda_qualified
from opendbc.car.subaru.stock_aol import qualified as subaru_qualified
from opendbc.car.hyundai.canfd_angle_aol import qualified as angle_qualified
@@ -466,7 +467,8 @@ class TransportPauseFeedback:
stock_user = stock_lateral_user_event(cp)
stock_temporary_user = bool((location_fault or eps_recovery) and (stock_hkg or chrysler_qualified(cp, marked_only=True) or
ford_qualified(cp, marked_only=True) or nissan_qualified(cp, marked_only=True)
or subaru_qualified(cp, marked_only=True) or mazda_qualified(cp, marked_only=True) or rivian_qualified(cp, marked_only=True)) and users and
or subaru_qualified(cp, marked_only=True) or mazda_qualified(cp, marked_only=True) or rivian_qualified(cp, marked_only=True)
or volkswagen_qualified(cp, marked_only=True)) and users and
not axis.desiredLongitudinal and not axis.longitudinalActive and
cs.cruiseState.available and not cs.cruiseState.enabled and
cs.gearShifter in (gears.drive, gears.low, gears.sport, gears.brake) and
@@ -726,7 +728,12 @@ class TransportPauseFeedback:
stock_rivian_eps = bool(rivian_temporary_eps_restriction(cp, cs) and
all(not event.userDisable or event.name == log.OnroadEvent.EventName.pcmDisable and
not event.softDisable and not event.immediateDisable for event in events))
restricted_eps = bool((temporary_eps_restriction(cp, cs) or stock_ford_eps or stock_mazda_eps or stock_rivian_eps) and self.lateral_only_baseline and
from openpilot.starpilot.car.volkswagen.aol import temporary_eps_restriction as volkswagen_temporary_eps_restriction
stock_volkswagen_eps = bool(volkswagen_temporary_eps_restriction(cp, cs) and
all(not event.userDisable or event.name == log.OnroadEvent.EventName.pcmDisable and
not event.softDisable and not event.immediateDisable for event in events))
restricted_eps = bool((temporary_eps_restriction(cp, cs) or stock_ford_eps or stock_mazda_eps or stock_rivian_eps or
stock_volkswagen_eps) and self.lateral_only_baseline and
not axis.desiredLongitudinal and not axis.longitudinalActive)
native = _configured_transport_native(sm, cp, self.session, now_ns, allow_restricted_lateral=restricted_eps)
if (native is None or not native.pandaInventory or
@@ -736,6 +743,8 @@ class TransportPauseFeedback:
native.observedMonoTime < self.lateral_native_ns or
(native.pandaSerial, tuple(p.hardwareSerial for p in native.pandaInventory)) != self.lateral_inventory):
return False
if stock_volkswagen_eps and matched_withdrawal:
self.volkswagen_eps_source_ns = int(axis.sourceCarStateMonoTime)
if cs.steerFaultTemporary and matched_withdrawal:
self.temporary_eps_event_ns = int(axis.faultEventMonoTime)
if pending_withdrawal:
@@ -1264,6 +1273,11 @@ class TransportPauseFeedback:
axis.desiredLateral and not axis.lateralActive):
self.failed = True
return True
if cp.brand == 'volkswagen' and current and continuity:
from openpilot.starpilot.car.volkswagen.aol import retain_eps_recovery
if retain_eps_recovery(self, sm, cp, cs, axis, now_ns, latched=latched, producer=intent_producer_session):
self.sequence, self.signature = int(axis.sequence), signature
return False
if current and continuity and self._ordinary_warning_intent_current(sm, cp, cs, axis, stamp, now_ns,
latched=latched, producer=intent_producer_session):
self.ordinary_warning_qualified = any(event.softDisable for event in sm['onroadEvents'])
+82 -3
View File
@@ -1,18 +1,19 @@
from opendbc.car import structs
from opendbc.car.volkswagen.stock_aol import qualified as stock_qualified, safety_config as aol_safety_config, AOL_WORD, MANUAL_WORD, DIRECT_WORD, AOL_WORDS
from opendbc.car.volkswagen.long_aol import qualified as long_qualified, WORDS as LONG_WORDS
from opendbc.car.volkswagen.meb_modes import qualified as meb_qualified, profile_word as meb_word, base_word as meb_base_word, AOL_WORDS as MEB_WORDS
from openpilot.starpilot.aol.intent import AolCardIntent, AOL_TOGGLE, ButtonType
from openpilot.starpilot.aol.policy import AolVehiclePolicy
def qualified(cp, *, marked_only=False):
return stock_qualified(cp, marked_only=marked_only) or long_qualified(cp, marked_only=marked_only)
return stock_qualified(cp, marked_only=marked_only) or long_qualified(cp, marked_only=marked_only) or meb_qualified(cp, marked_only=marked_only)
def temporary_restriction(cp, state):
gears = structs.CarState.GearShifter
return bool(state is not None and qualified(cp, marked_only=True) and state.canValid and not state.canTimeout and
not state.steerFaultPermanent and (state.steerFaultTemporary or state.standstill or
not state.steerFaultPermanent and (state.steerFaultTemporary or (state.standstill and not cp.steerAtStandstill) or
state.gearShifter not in (gears.drive, gears.sport, gears.eco, gears.manumatic)))
@@ -21,6 +22,8 @@ def policy_for(cp):
return AolVehiclePolicy()
word = AOL_WORD | (MANUAL_WORD if cp.transmissionType == structs.CarParams.TransmissionType.manual else 0)
word |= DIRECT_WORD if cp.transmissionType == structs.CarParams.TransmissionType.direct else 0
if meb_qualified(cp):
word = meb_word(cp) & ~meb_base_word(cp)
return AolVehiclePolicy(intent_supported=True, settings_supported=True, runtime_supported=True,
normal_runtime_supported=True, ordinary_axis_ack_required=qualified(cp, marked_only=True),
explicit_latch=True, lkas_button_supported=False, safety_param_addition=word,
@@ -28,7 +31,8 @@ def policy_for(cp):
def native_profile_supported(model, param):
return model == int(structs.CarParams.SafetyModel.volkswagen) and param in AOL_WORDS | LONG_WORDS
return (model == int(structs.CarParams.SafetyModel.volkswagen) and param in AOL_WORDS | LONG_WORDS or
model == int(structs.CarParams.SafetyModel.volkswagenMeb) and param in MEB_WORDS)
def native_accepts_cp(cp, model, param):
@@ -51,6 +55,9 @@ class VolkswagenStockCardIntent(AolCardIntent):
observe_stock_engagement = True
def __init__(self, cp, settings):
self.cp = cp
self.physical_eps_pause = False
self.eps_recovery_ns = 0
self.main_seen = False
self.main_startup_pending = True
self.main_available = False
@@ -59,6 +66,21 @@ class VolkswagenStockCardIntent(AolCardIntent):
super().__init__(settings, explicit_latch=True, lkas_button_supported=False)
def update(self, CS, **kwargs):
from openpilot.starpilot.aol.runtime import SAFETY_MAX_AGE_NS
now_ns = kwargs.get('now_ns', 0)
if temporary_eps_restriction(self.cp, CS):
self.physical_eps_pause = True
self.eps_recovery_ns = 0
elif self.physical_eps_pause and _eps_physical_state(self.cp, CS):
self.physical_eps_pause = False
self.eps_recovery_ns = now_ns
elif not _eps_physical_state(self.cp, CS):
self.physical_eps_pause = False
self.eps_recovery_ns = 0
rejection_ns = kwargs.get('native_rejection_ns', 0)
if (kwargs.get('fault_active') is False and 0 < rejection_ns <= self.eps_recovery_ns <= now_ns and
now_ns - self.eps_recovery_ns <= SAFETY_MAX_AGE_NS):
kwargs['native_rejection_ns'] = 0
state = structs.CarState(**CS.to_dict())
events = []
for event in CS.buttonEvents:
@@ -104,3 +126,60 @@ class VolkswagenStockCardIntent(AolCardIntent):
def create_intent(cp, settings):
return VolkswagenStockCardIntent(cp, settings)
def _eps_physical_state(cp, state):
try:
gears = structs.CarState.GearShifter
return bool(qualified(cp, marked_only=True) and state is not None and state.canValid and not state.canTimeout and
not state.steerFaultPermanent and state.cruiseState.available and not state.cruiseState.enabled and
state.gearShifter in (gears.drive, gears.sport, gears.eco, gears.manumatic) and
not state.brakePressed and not state.regenBraking and not state.gasPressed and not state.doorOpen and
not state.seatbeltUnlatched and not state.accFaulted and not state.vehicleSensorsInvalid and
not state.steeringDisengage and not state.buttonEvents)
except (AttributeError, KeyError, IndexError, TypeError, ValueError, OverflowError):
return False
def temporary_eps_restriction(cp, state):
"""Exact already armed LAT-only EPS withdrawal; never output permission."""
return bool(_eps_physical_state(cp, state) and state.steerFaultTemporary)
def retain_eps_recovery(feedback, sm, cp, state, axis, now_ns, *, latched, producer):
"""Join only an older EPS denial to its verified physical recovery; grant no axis."""
from openpilot.cereal import log
from openpilot.starpilot.aol.runtime import AXIS_MAX_AGE_NS, SAFETY_MAX_AGE_NS
from openpilot.starpilot.aol.transport_pause import _configured_transport_native
try:
last = int(getattr(feedback, 'volkswagen_eps_source_ns', 0))
stamp = int(sm.logMonoTime['aolAxisState'])
if (not latched or not _eps_physical_state(cp, state) or state.steerFaultTemporary or
feedback.failed or feedback.paused or feedback.pending_event or feedback.retention_started_ns or
not feedback.temporary_eps_event_ns or not feedback.lateral_only_baseline or
not feedback.lateral_baseline_ns or not feedback.lateral_inventory or
not producer or producer != feedback.availability_producer or
axis.intentProducerSessionId != producer or axis.intentIntegrityEpoch != feedback.availability_epoch or
not feedback.session or str(axis.sessionId) != feedback.session or axis.faultSessionId != axis.sessionId or
str(axis.faultReason) != 'none' or str(axis.mode) not in ('off', 'lateralOnly') or
axis.lateralActive or axis.desiredLongitudinal or axis.longitudinalActive or
not 0 < last <= axis.sourceCarStateMonoTime <= stamp == axis.observedMonoTime <= now_ns <= axis.validUntilMonoTime or
now_ns - last > SAFETY_MAX_AGE_NS or now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
axis.validUntilMonoTime - axis.observedMonoTime > AXIS_MAX_AGE_NS or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents']):
return False
event_ns = int(sm.logMonoTime['onroadEvents'])
temporary = (log.OnroadEvent.EventName.steerTempUnavailable, log.OnroadEvent.EventName.steerTempUnavailableSilent)
if (not 0 < feedback.temporary_eps_event_ns <= event_ns <= now_ns or now_ns - event_ns > 1_500_000_000 or
any(event.immediateDisable or event.softDisable and event.name not in temporary or
event.userDisable and not (cp.pcmCruise and event.name == log.OnroadEvent.EventName.pcmDisable)
for event in sm['onroadEvents'])):
return False
native = _configured_transport_native(sm, cp, feedback.session, now_ns,
require_longitudinal=False, allow_restricted_lateral=True)
return bool(native is not None and native.pandaInventory and
feedback.lateral_native_ns <= native.observedMonoTime <= last and
not native.lateralAllowed and not native.requestedLongitudinal and not native.longitudinalAllowed and
(native.pandaSerial, tuple(p.hardwareSerial for p in native.pandaInventory)) == feedback.lateral_inventory)
except (AttributeError, KeyError, IndexError, TypeError, ValueError, OverflowError):
return False
@@ -8,3 +8,9 @@ inline bool volkswagen_long_aol_param(uint16_t param) {
}
inline constexpr AolSafetyProfile VOLKSWAGEN_LONG_AOL_PROFILE{15U, volkswagen_long_aol_param, true};
inline constexpr AolSafetyProfile VOLKSWAGEN_STOCK_AOL_PROFILE{15U, volkswagen_stock_aol_param, true};
inline bool volkswagen_meb_aol_param(uint16_t param) {
return (param == 8U) || (param == 10U) || (param == 12U) || (param == 14U) ||
(param == 24U) || (param == 26U) || (param == 28U) || (param == 30U);
}
inline constexpr AolSafetyProfile VOLKSWAGEN_MEB_AOL_PROFILE{34U, volkswagen_meb_aol_param, true};
@@ -0,0 +1,10 @@
"""A saved OFF remains OFF; an absent legacy Alpha key keeps upstream MEB defaults."""
from openpilot.starpilot.saved_source import read_saved
def stock_long_requested(params):
try:
raw, readable = read_saved(params, 'AlphaLongitudinalEnabled', 8)
return not readable or raw not in (None, b'1')
except (OSError, TypeError, ValueError):
return True
@@ -298,7 +298,8 @@ class FeatureSettingsOwner:
from opendbc.car.gm.startup_preferences import disable_long_supported
cp = self.vehicle_params()
try:
if cp is None or not (bolt_disable_supported(cp) or disable_long_supported(cp)):
from opendbc.car.volkswagen.meb_modes import supported as meb_stock_supported
if cp is None or not (bolt_disable_supported(cp) or disable_long_supported(cp) or meb_stock_supported(cp)):
return None
return (str(cp.carFingerprint), str(cp.brand), bool(cp.openpilotLongitudinalControl),
bool(cp.pcmCruise), int(cp.flags),
@@ -306,12 +307,18 @@ class FeatureSettingsOwner:
except (AttributeError, TypeError, ValueError, OverflowError):
return None
def _vw_alpha_off(self) -> bool:
from opendbc.car.volkswagen.meb_modes import supported as meb_stock_supported
from openpilot.starpilot.car.volkswagen.preferences import stock_long_requested
return meb_stock_supported(self.vehicle_params()) and stock_long_requested(self.params)
def _apply_bolt_disable(self, request: FeatureSettingsRequest) -> bool:
if not request.confirmation or request.value not in ("Off", "On") or request.dependencies:
return False
def authorized() -> bool:
return (bool(request.vehicle_fingerprint) and self.vehicle_fingerprint() == request.vehicle_fingerprint and
request.capability is not None and request.capability == self._bolt_disable_capability() and
not (request.value == "Off" and self._vw_alpha_off()) and
self.authority("parked_preferences") and request.expected in (None, b"0", b"1"))
return commit_exact(self.params, key="DisableOpenpilotLongitudinal", max_bytes=8,
raw=b"1" if request.value == "On" else b"0", expected=request.expected,
@@ -871,6 +878,9 @@ class FeatureSettingsOwner:
row = self._bool_row("DisableOpenpilotLongitudinal", "Disable StarPilot Speed Control",
parked and self.authority("parked_preferences"),
"StarPilot does not control speed in this mode. Steering remains available after the next startup")
if self._vw_alpha_off():
row = replace(row, value="On", available=False,
reason="Speed control is off in driving settings. Enable Alpha Longitudinal there to restore it")
if bolt_capability[-1] == (("noOutput", 0),):
row = replace(row, reason="Factory camera not detected. Turn this setting off and restart to restore StarPilot speed control.")
rows.append(replace(row, available=row.available and self._readable("DisableOpenpilotLongitudinal"),
@@ -0,0 +1,105 @@
"""Both native settings surfaces retain and expose the exact release-supported MEB option."""
from tempfile import TemporaryDirectory
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
import pytest
from opendbc.car.volkswagen.interface import CarInterface
from opendbc.car.volkswagen.values import CAR
from opendbc.car.volkswagen.meb_modes import prepare_stock
from openpilot.common.params import Params
from openpilot.selfdrive.ui.layouts.settings import developer as large
from openpilot.selfdrive.ui.mici.layouts.settings import developer as compact
from openpilot.starpilot.ui.tests.test_ioniq6_alpha_request import Toggle
@pytest.mark.parametrize('module,layout', ((large, large.DeveloperLayout), (compact, compact.DeveloperLayoutMici)))
@pytest.mark.parametrize('release', (False, True))
@pytest.mark.parametrize('raw,expected', ((None, True), (b'0', False), (b'1', True), (b'bad', False)))
def test_saved_meb_off_can_be_restored_on_each_native_surface(module, layout, release, raw, expected):
with TemporaryDirectory() as directory:
params = Params(directory)
path = Path(params.get_param_path('AlphaLongitudinalEnabled'))
if raw is not None:
path.write_bytes(raw)
fp = {bus: {} for bus in range(8)}
fp[1][0x13D] = 32
cp = CarInterface.get_params(CAR.VOLKSWAGEN_ID4_MK1, fp, [], False, release, False)
prepare_stock(cp, True)
ui = SimpleNamespace(CP=cp, params=params, is_release=release, engaged=False,
has_longitudinal_control=False, is_offroad=lambda: True, update_params=lambda: None)
panel = layout.__new__(layout)
panel._alpha_long_toggle = Toggle()
panel._long_maneuver_toggle = Toggle()
panel._lat_maneuver_toggle = Toggle()
if layout is large.DeveloperLayout:
panel._params = params
panel._is_release = release
panel._joystick_toggle = Toggle()
panel._adb_toggle = Toggle()
panel._ssh_toggle = Toggle()
panel._ui_debug_toggle = Toggle()
panel._cellular_disabled_toggle = Toggle()
else:
panel._refresh_toggles = (('AlphaLongitudinalEnabled', panel._alpha_long_toggle),)
with patch.object(module, 'ui_state', ui):
panel._update_toggles()
assert panel._alpha_long_toggle.visible
assert panel._alpha_long_toggle.state == expected
assert (path.read_bytes() if path.exists() else None) == raw
cp.passive = True
cp.dashcamOnly = True
panel._update_toggles()
assert not panel._alpha_long_toggle.visible
assert (path.read_bytes() if path.exists() else None) == raw
@pytest.mark.parametrize('module,layout', ((large, large.DeveloperLayout), (compact, compact.DeveloperLayoutMici)))
def test_release_meb_confirmation_rechecks_offroad_and_saves_actual_choice(module, layout):
with TemporaryDirectory() as directory:
params = Params(directory)
params.put_bool('AlphaLongitudinalEnabled', False, block=True)
fp = {bus: {} for bus in range(8)}
fp[1][0x13D] = 32
cp = CarInterface.get_params(CAR.VOLKSWAGEN_ID4_MK1, fp, [], False, True, False)
prepare_stock(cp, True)
parked = [True]
ui = SimpleNamespace(CP=cp, params=params, is_release=True, engaged=False,
has_longitudinal_control=False, is_offroad=lambda: parked[0], update_params=lambda: None)
class AlphaToggle(Toggle):
title = 'Speed control'
description = 'Use StarPilot speed control.'
panel = layout.__new__(layout)
panel._alpha_long_toggle = AlphaToggle()
panel._long_maneuver_toggle = Toggle()
panel._lat_maneuver_toggle = Toggle()
if layout is large.DeveloperLayout:
panel._params = params
panel._is_release = True
panel._joystick_toggle = Toggle()
panel._adb_toggle = Toggle()
panel._ssh_toggle = Toggle()
panel._ui_debug_toggle = Toggle()
panel._cellular_disabled_toggle = Toggle()
else:
panel._refresh_toggles = (('AlphaLongitudinalEnabled', panel._alpha_long_toggle),)
confirm = []
def large_dialog(*args, callback, **kwargs):
confirm.append(lambda: callback(large.DialogResult.CONFIRM))
return object()
def compact_dialog(callback, *, system_long_default=False):
assert system_long_default
confirm.append(callback)
return object()
with (patch.object(module, 'ui_state', ui), patch.object(module.gui_app, 'push_widget'),
patch.object(large, 'ConfirmDialog', side_effect=large_dialog),
patch.object(compact, 'AlphaLongConfirmPage', side_effect=compact_dialog),
patch.object(compact, 'restart_needed_callback')):
panel._on_alpha_long_enabled(True)
assert len(confirm) == 1 and not params.get_bool('AlphaLongitudinalEnabled')
parked[0] = False
confirm[0]()
assert not params.get_bool('AlphaLongitudinalEnabled')
parked[0] = True
confirm[0]()
assert params.get_bool('AlphaLongitudinalEnabled')
+11 -2
View File
@@ -46,6 +46,7 @@ class VehicleStartupPreferences:
toyota_unlock_doors: bool = False
toyota_radar_mode: int = 0
gm_cancel_remap: bool = False
vw_stock_long: bool = field(default=False, kw_only=True)
@classmethod
def read(cls, params, *, enabled: bool):
@@ -59,6 +60,8 @@ class VehicleStartupPreferences:
radar = read_radar_settings(params)
pedal = read_pedal_settings(params)
preap_stock = stock_configuration(params, radar, pedal)
from openpilot.starpilot.car.volkswagen.preferences import stock_long_requested
vw_stock_long = stock_long_requested(params)
try:
raw_radar, radar_readable = read_saved(params, "HondaBoschARadar", 8)
honda_radar = radar_readable and raw_radar in (None, b"1")
@@ -82,7 +85,8 @@ class VehicleStartupPreferences:
volt_one_pedal = bool(enabled and not disable_bolt and read_saved(params, "VoltOnePedalMode", 8) == (b"1", True) and
readable and safe in (None, b"0"))
except (OSError, TypeError, ValueError):
return cls(subaru_smoothing=smoothing, disable_bolt_long=disable_bolt, honda_bosch_a_radar=honda_radar, tesla_preap_stock=preap_stock,
return cls(vw_stock_long=vw_stock_long,
subaru_smoothing=smoothing, disable_bolt_long=disable_bolt, honda_bosch_a_radar=honda_radar, tesla_preap_stock=preap_stock,
tesla_preap_radar=radar, tesla_preap_pedal=pedal,
gm_camera_pedal=False, toyota_filter=False)
try:
@@ -99,7 +103,8 @@ class VehicleStartupPreferences:
read_saved(params, 'AlwaysOnLateral', 8) == (b'1', True) and
read_saved(params, 'TeslaAOLScreenTap', 8) == (b'1', True))
screen_brake = read_saved(params, 'TeslaAOLDisengageOnBrake', 8) == (b'1', True)
return cls(subaru_smoothing=smoothing, subaru_avh_startup=bool(enabled and readable and safe in (None, b"0") and
return cls(vw_stock_long=vw_stock_long,
subaru_smoothing=smoothing, subaru_avh_startup=bool(enabled and readable and safe in (None, b"0") and
read_saved(params, "IsReleaseBranch", 8) in ((None, True), (b"0", True)) and
read_saved(params, "SubaruAvhStartup", 8) == (b"1", True)),
subaru_stop_start_off=bool(enabled and readable and safe in (None, b"0") and
@@ -170,6 +175,8 @@ class VehicleStartupPreferences:
prepare_disable_longitudinal(cp, True)
def prepare(self, cp, *, fingerprints=None):
from opendbc.car.volkswagen.meb_modes import prepare_stock as prepare_meb_stock
prepare_meb_stock(cp, self.disable_bolt_long or self.vw_stock_long)
from opendbc.car.subaru.software_cruise import prepare
cp = prepare(cp, self.subaru_software_cruise)
from opendbc.car.subaru.sng import prepare as prepare_sng
@@ -215,6 +222,8 @@ class VehicleStartupPreferences:
return cp
def finalize(self, cp) -> None:
from opendbc.car.volkswagen.meb_modes import prepare_stock as prepare_meb_stock
prepare_meb_stock(cp, self.disable_bolt_long or self.vw_stock_long)
from opendbc.car.gm.values import apply_gm_auto_hold, is_gm_auto_hold, apply_volt_one_pedal, is_volt_one_pedal
admitted_one_pedal = is_volt_one_pedal(cp)
admitted_hold = is_gm_auto_hold(cp)
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+1 -1
View File
@@ -1,4 +1,4 @@
#pragma once
extern const uint8_t gitversion[29];
const uint8_t gitversion[29] = "DEV-4e956b46-Pa5919516-DEBUG";
const uint8_t gitversion[29] = "DEV-b98f1bfc-Pa5919516-DEBUG";
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+1 -1
View File
@@ -1 +1 @@
{"protected_inventory_sha256": "8524cc5d7574ac2b6ecfc6d9eb30d0e36becb8fa4e996bba456e6ee1870ec249", "schema_version": 1}
{"protected_inventory_sha256": "1ce92f195f4b5461eac417896c24994bfcd601e29f2738fc499b57987af56687", "schema_version": 1}
+7
View File
@@ -77,6 +77,11 @@ AOL_LIFECYCLE_TARGETS = (
"openpilot/starpilot/aol/tests/test_volkswagen.py",
"openpilot/starpilot/aol/tests/test_volkswagen_saved_intent.py",
"openpilot/starpilot/aol/tests/test_mqb_actual_callers.py",
"openpilot/starpilot/aol/tests/test_meb_actual_callers.py",
"opendbc_repo/opendbc/car/volkswagen/tests/test_meb_preferences.py",
"opendbc_repo/opendbc/car/volkswagen/tests/test_meb_discovery.py",
"opendbc_repo/opendbc/car/volkswagen/tests/test_meb_stock_modes.py",
"openpilot/starpilot/ui/tests/test_meb_preferences.py",
"openpilot/starpilot/aol/tests/test_mqb_direct_callers.py",
"openpilot/starpilot/aol/tests/test_mqb_topology_callers.py",
"openpilot/starpilot/aol/tests/test_mqb_topology_contract.py",
@@ -85,6 +90,7 @@ AOL_LIFECYCLE_TARGETS = (
"opendbc_repo/opendbc/safety/tests/test_volkswagen_stock_aol.py",
"opendbc_repo/opendbc/safety/tests/test_volkswagen_engagement_arm.py",
"opendbc_repo/opendbc/safety/tests/test_volkswagen_long_aol.py",
"opendbc_repo/opendbc/safety/tests/test_meb_aol.py",
"opendbc_repo/opendbc/safety/tests/test_volkswagen_direct_aol.py",
"openpilot/starpilot/aol/tests/test_volkswagen_main_boot.py",
"openpilot/starpilot/car/subaru/tests/test_aol.py",
@@ -315,6 +321,7 @@ RELEASE_TARGETS = (
"opendbc_repo/opendbc/safety/tests/test_volkswagen_stock_aol.py",
"opendbc_repo/opendbc/safety/tests/test_volkswagen_engagement_arm.py",
"opendbc_repo/opendbc/safety/tests/test_volkswagen_long_aol.py",
"opendbc_repo/opendbc/safety/tests/test_meb_aol.py",
"opendbc_repo/opendbc/safety/tests/test_volkswagen_direct_aol.py",
"openpilot/starpilot/aol/tests/test_volkswagen_main_boot.py",
"opendbc_repo/opendbc/safety/tests/test_mazda.py::TestMazdaSafety",