Support original Prius four-byte CAN filters

This commit is contained in:
firestar5683
2026-10-23 12:17:52 -05:00
parent 7df9af125d
commit 9d2d8f3a31
21 changed files with 635 additions and 45 deletions
@@ -9,7 +9,7 @@ from opendbc.car.secoc import add_mac, build_sync_mac
from opendbc.car.interfaces import CarControllerBase
from opendbc.car.toyota import toyotacan
from opendbc.car.toyota import prius_longitudinal, rav4h_filter, pedal, eps8_sdsu, legacy_sdsu, advert_filter, legacy_stop_timer, legacy_advert_filter
from opendbc.car.toyota import camera_2aa_filter, prius_sdsu
from opendbc.car.toyota import camera_2aa_filter, prius_sdsu, prius_advert_filter
from opendbc.car.toyota.door_locks import DoorLocks
from opendbc.car.toyota.values import CAR, CarControllerParams, ToyotaFlags, TOYOTA_AUTO_HOLD_CARS, uses_toyota_auto_hold_aeb
from opendbc.can import CANPacker
@@ -75,6 +75,7 @@ class CarController(CarControllerBase):
self.reverse_cruise_input = None
self.prius_filter_input = None
self.original_prius_sdsu = prius_sdsu.configured(CP)
self.original_prius_filter = self.original_prius_sdsu or prius_advert_filter.configured(CP)
self.pedal_input = None
self.rav4h_filter_input = None
self.advert_filter_input = None
@@ -146,10 +147,10 @@ class CarController(CarControllerBase):
_projection.begin(CC.actuators.torque, self.frame)
filter_allowed = (not self.prius_longitudinal or self.prius_filter_input is not None and
self.prius_filter_input.update(now_nanos))
original_prius_long = self.original_prius_sdsu and self.CP.openpilotLongitudinalControl
if self.original_prius_sdsu:
original_prius_long = self.original_prius_filter and self.CP.openpilotLongitudinalControl
if self.original_prius_filter:
filter_allowed = filter_allowed and bool(
CS.out.canValid and not CS.out.canTimeout and CS.out.cruiseState.available and CS.sdsu_state == 0 and
CS.out.canValid and not CS.out.canTimeout and CS.out.cruiseState.available and (not self.original_prius_sdsu or CS.sdsu_state == 0) and
not CS.out.accFaulted and not CS.out.stockAeb and not CS.out.steerFaultTemporary and not CS.out.steerFaultPermanent and
CS.out.gearShifter in (structs.CarState.GearShifter.drive, structs.CarState.GearShifter.sport) and not CS.out.doorOpen and
not CS.out.seatbeltUnlatched and not CS.out.gasPressed and not CS.out.brakePressed)
@@ -186,7 +187,7 @@ class CarController(CarControllerBase):
if (rav4h_long or pedal_owner or advert_long or original_prius_long) and not filter_allowed and CS.out.cruiseState.enabled:
pcm_cancel_cmd = True
lat_active = CC.latActive and abs(CS.out.steeringTorque) < MAX_USER_TORQUE
if self.original_prius_sdsu:
if self.original_prius_filter:
lat_active = lat_active and CS.out.canValid and not CS.out.canTimeout and CS.out.cruiseState.available and not (
CS.out.accFaulted or CS.out.steerFaultTemporary or CS.out.steerFaultPermanent or CS.out.doorOpen)
if advert_owner:
+10 -8
View File
@@ -2,7 +2,7 @@ import copy
import math
from opendbc.can import CANDefine, CANParser
from opendbc.car.toyota import pedal, legacy_sdsu, advert_filter, legacy_advert_filter, prius_sdsu
from opendbc.car.toyota import pedal, legacy_sdsu, advert_filter, legacy_advert_filter, prius_sdsu, prius_advert_filter
from opendbc.car.toyota.lexus_is import LexusIsCANParser
from opendbc.car import Bus, DT_CTRL, create_button_events, structs
from opendbc.car.common.conversions import Conversions as CV
@@ -33,7 +33,8 @@ class CarState(CarStateBase):
def __init__(self, CP):
super().__init__(CP)
can_define = CANDefine(DBC[CP.carFingerprint][Bus.pt])
self.eps_torque_scale = (.73 if prius_filter_configured(CP) and not prius_sdsu.configured(CP) else EPS_SCALE[CP.carFingerprint] / 100.)
original_prius = prius_sdsu.configured(CP) or prius_advert_filter.configured(CP)
self.eps_torque_scale = (.73 if prius_filter_configured(CP) and not original_prius else EPS_SCALE[CP.carFingerprint] / 100.)
self.cluster_speed_hyst_gap = CV.KPH_TO_MS / 2.
self.cluster_min_speed = CV.KPH_TO_MS / 2.
@@ -244,7 +245,7 @@ class CarState(CarStateBase):
if (self.CP.carFingerprint in (CAR.TOYOTA_PRIUS, CAR.TOYOTA_PRIUS_RETROFIT) or
self.CP.flags & ToyotaFlags.LONG_FILTER or advert_filter.configured(self.CP)):
prev_distance_button = self.distance_button
if advert_filter.configured(self.CP):
if advert_filter.configured(self.CP) or prius_advert_filter.configured(self.CP):
fresh = cp.advert_button_ns > 0 and 0 <= cp._last_update_nanos - cp.advert_button_ns <= 200_000_000
self.distance_button = cp.advert_distance_button if fresh else 0
else:
@@ -285,7 +286,7 @@ class CarState(CarStateBase):
sdsu_fresh = source_ns > 0 and source_ns <= cp._last_update_nanos <= valid_until_ns
ret.accFaulted = ret.accFaulted or not sdsu_fresh or self.sdsu_state not in (0, 7)
ret.stockAeb = ret.stockAeb or self.sdsu_state == 7
if original_prius_sdsu:
if original_prius_sdsu or prius_advert_filter.configured(self.CP):
main_current = cp.main_valid and prius_sdsu.source_current(cp, 'PCM_CRUISE_2', 'MAIN_ON')
eps_current = cp.eps_valid and prius_sdsu.source_current(cp, 'EPS_STATUS', 'LKA_STATE')
ret.cruiseState.available = ret.cruiseState.available and main_current and not cp.hardware_ambiguous
@@ -325,18 +326,19 @@ class CarState(CarStateBase):
if CP.carFingerprint == CAR.LEXUS_IS or alternate_sdsu_main:
pt_messages.append(("DSU_CRUISE", 10))
if (advert_filter.configured(CP) and not alternate_sdsu_main) or prius_sdsu.configured(CP):
if (advert_filter.configured(CP) and not alternate_sdsu_main) or prius_sdsu.configured(CP) or prius_advert_filter.configured(CP):
pt_messages.append(("PCM_CRUISE_2", 10))
if prius_sdsu.configured(CP):
if prius_sdsu.configured(CP) or prius_advert_filter.configured(CP):
pt_messages.append(("EPS_STATUS", 25))
if CP.flags & ToyotaFlags.LONG_FILTER:
if CP.flags & ToyotaFlags.LONG_FILTER and not prius_advert_filter.configured(CP):
pt_messages.append(("SDSU", 50))
if CP.flags & ToyotaFlags.ZSS:
pt_messages.append(("SECONDARY_STEER_ANGLE", math.nan))
if pedal.configured(CP):
pt_messages.append(("GAS_SENSOR", 50))
pt_parser = (prius_sdsu.PriusSDSUCANParser if prius_sdsu.configured(CP) else
pt_parser = (prius_advert_filter.PriusAdvertCANParser if prius_advert_filter.configured(CP) else
prius_sdsu.PriusSDSUCANParser if prius_sdsu.configured(CP) else
camera_2aa_filter.CameraTwoAACANParser if camera_twoaa else
legacy_advert_filter.LegacyAdvertCANParser if legacy_advert else
legacy_sdsu.AlternateMainSDSUCANParser if alternate_sdsu_main else
@@ -239,6 +239,9 @@ class CarInterface(CarInterfaceBase):
from opendbc.car.toyota import prius_sdsu
if prius_sdsu.hardware_supported(candidate, fingerprint, car_fw):
prius_sdsu.apply(ret)
from opendbc.car.toyota import prius_advert_filter
if prius_advert_filter.hardware_supported(candidate, fingerprint, car_fw):
prius_advert_filter.apply(ret)
from opendbc.car.toyota import pedal
if pedal.hardware_supported(candidate, fingerprint, car_fw):
ret.flags |= int(ToyotaFlags.GAS_INTERCEPTOR)
@@ -0,0 +1,75 @@
"""Original EPS66 Prius four-byte filter; no smartDSU status is inferred."""
from opendbc.car import structs
from opendbc.car.toyota.advert_filter import AdvertFilterCANParser
from opendbc.car.toyota.values import CAR, ToyotaFlags
LONG_WORD = 0xC042
STOCK_WORD = 0xC242
WORDS = frozenset((LONG_WORD, STOCK_WORD))
RETROFIT_EPS = b'8965B47070\x00\x00\x00\x00\x00\x00'
def hardware_supported(candidate, fingerprint, car_fw):
if (candidate != CAR.TOYOTA_PRIUS or fingerprint.get(0, {}).get(0x2FF) != 4 or
any(0x2FF in messages and (bus != 2 or messages[0x2FF] != 4) for bus, messages in fingerprint.items() if bus != 0) or
any(address in messages for messages in fingerprint.values() for address in (0x201, 0x2AA)) or
any(fingerprint.get(0, {}).get(address, 8) != 8 for address in (0x1D3, 0x262)) or
any(fingerprint.get(2, {}).get(address, 8) != 8 for address in (0x343, 0x4CB))):
return False
from opendbc.car.toyota.fingerprints import FW_VERSIONS
known = FW_VERSIONS[candidate]
required = {structs.CarParams.Ecu.eps, structs.CarParams.Ecu.fwdCamera}
if any(ecu == structs.CarParams.Ecu.fwdRadar for ecu, _, _ in known):
required.add(structs.CarParams.Ecu.fwdRadar)
for ecu in required:
observed = [fw for fw in car_fw if fw.ecu == ecu]
if not observed or any(fw.logging or fw.bus != 0 or fw.brand != 'toyota' or
bytes(fw.fwVersion) not in known.get((ecu, fw.address, fw.subAddress or None), ()) or
ecu == structs.CarParams.Ecu.eps and bytes(fw.fwVersion) == RETROFIT_EPS for fw in observed):
return False
return True
def apply(cp):
cp.flags |= int(ToyotaFlags.HYBRID | ToyotaFlags.LONG_FILTER | ToyotaFlags.RAISED_ACCEL_LIMIT)
cp.safetyConfigs[0].safetyParam = LONG_WORD
cp.openpilotLongitudinalControl = True
cp.alphaLongitudinalAvailable = False
def configured(cp):
if (cp is None or cp.brand != 'toyota' or cp.carFingerprint != CAR.TOYOTA_PRIUS or cp.notCar or cp.passive or
cp.dashcamOnly or cp.secOcRequired or not cp.pcmCruise or len(cp.safetyConfigs) != 1 or
cp.steerControlType != structs.CarParams.SteerControlType.torque):
return False
flags = int(ToyotaFlags.HYBRID | ToyotaFlags.LONG_FILTER | ToyotaFlags.RAISED_ACCEL_LIMIT)
config = cp.safetyConfigs[0]
return bool(cp.flags & ~int(ToyotaFlags.ZSS) == flags and
config.safetyModel == structs.CarParams.SafetyModel.toyota and
config.safetyParam == (LONG_WORD if cp.openpilotLongitudinalControl else STOCK_WORD))
def prepare_stock(cp):
if not configured(cp):
return False
cp.openpilotLongitudinalControl = False
cp.safetyConfigs[0].safetyParam = STOCK_WORD
cp.autoResumeSng = False
return True
class PriusAdvertCANParser(AdvertFilterCANParser):
"""The discriminator carries FD_BUTTON only; MAIN/EPS retain authority."""
def __init__(self, *args):
super().__init__(*args)
self.hardware_ambiguous = False
def update(self, strings, sendcan=False):
batches = []
for stamp, frames in strings:
frames = list(frames)
for address, data, bus in frames:
if address in (0x201, 0x2AA) or address == 0x2FF and (bus not in (0, 2) or len(data) != 4):
self.hardware_ambiguous = True
batches.append((stamp, frames))
return super().update(batches, sendcan)
@@ -13,8 +13,8 @@ TOYOTA_NO_LEAD_CRUISE_SIGN_FLIP_MIN_SET_SPEED_ERROR = .35
def configured(cp):
from opendbc.car.toyota import prius_sdsu
if prius_sdsu.configured(cp):
from opendbc.car.toyota import prius_sdsu, prius_advert_filter
if prius_sdsu.configured(cp) or prius_advert_filter.configured(cp):
return True
return (cp.brand == 'toyota' and cp.carFingerprint in (CAR.TOYOTA_PRIUS, CAR.TOYOTA_PRIUS_RETROFIT) and
cp.pcmCruise and not cp.dashcamOnly and not cp.notCar and
@@ -34,8 +34,8 @@ def stopping_decel_rate(cp):
def prepare_stock(cp):
from opendbc.car.toyota import prius_sdsu
if prius_sdsu.prepare_stock(cp):
from opendbc.car.toyota import prius_sdsu, prius_advert_filter
if prius_sdsu.prepare_stock(cp) or prius_advert_filter.prepare_stock(cp):
return True
if not configured(cp):
return False
+14 -10
View File
@@ -110,12 +110,12 @@ static bool toyota_legacy_main_current(void) {
static bool toyota_legacy_advert_ready(void) {
bool ready = true;
if (toyota_legacy_advert() || toyota_camera_2aa() || toyota_prius_sdsu()) {
if (toyota_legacy_advert() || toyota_camera_2aa() || toyota_prius_filter()) {
static const uint32_t timeout[4] = {303031U, 10000000U, 3333334U, 400000U};
const uint32_t now = microsecond_timer_get();
ready = aol_rx_healthy() && toyota_aol_main && toyota_aol_drive && toyota_aol_doors_closed && toyota_aol_eps &&
(!toyota_camera_2aa() || !toyota_camera_2aa_ambiguous) &&
(!toyota_prius_sdsu() || !toyota_prius_sdsu_ambiguous);
(!toyota_prius_filter() || !toyota_prius_sdsu_ambiguous);
for (unsigned int i = 0U; i < 4U; i++) {
ready &= toyota_aol_seen[i] && (safety_get_ts_elapsed(now, toyota_aol_ts[i]) <= timeout[i]);
}
@@ -193,6 +193,10 @@ static bool toyota_get_quality_flag_valid(const CANPacket_t *msg) {
}
static void toyota_optional_rx_hook(const CANPacket_t *msg) {
if (toyota_prius_advert() && ((msg->addr == 0x201U) || (msg->addr == 0x2AAU) ||
((msg->addr == 0x2FFU) && (((msg->bus != 0U) && (msg->bus != 2U)) || (GET_LEN(msg) != 4U))))) {
toyota_prius_sdsu_ambiguous = true;
}
if (toyota_prius_sdsu() && ((msg->addr == 0x201U) || (msg->addr == 0x2AAU) ||
((msg->addr == 0x2FFU) && (((msg->bus != 0U) && (msg->bus != 2U)) || (GET_LEN(msg) == 4U))))) {
toyota_prius_sdsu_ambiguous = true;
@@ -207,7 +211,7 @@ static void toyota_optional_rx_hook(const CANPacket_t *msg) {
const unsigned int aol_source_len = ((toyota_aol_rav4h_filter || toyota_alt_cruise || toyota_legacy_sdsu || toyota_legacy_advert()) && (msg->addr == 0x262U)) ? 5U : 8U;
if (!aol_checksum_source || ((GET_LEN(msg) == aol_source_len) && (toyota_get_checksum(msg) == toyota_compute_checksum(msg)))) {
toyota_aol_observe(msg);
} else if ((toyota_aol_enabled || toyota_legacy_advert() || toyota_camera_2aa() || toyota_prius_sdsu()) && (msg->bus == 0U)) {
} else if ((toyota_aol_enabled || toyota_legacy_advert() || toyota_camera_2aa() || toyota_prius_filter()) && (msg->bus == 0U)) {
unsigned int index = 3U;
if (msg->addr == 0x1D3U) {
index = 0U;
@@ -242,7 +246,7 @@ static void toyota_optional_rx_hook(const CANPacket_t *msg) {
static void toyota_rx_hook(const CANPacket_t *msg) {
// Required sources have already passed RX length/checksum validation here.
// Optional ingress only runs for messages outside the required RX catalog.
if ((toyota_legacy_advert() || toyota_camera_2aa() || toyota_prius_sdsu()) && ((msg->addr == 0x262U) || (!toyota_alt_cruise && (msg->addr == 0x1D3U)))) {
if ((toyota_legacy_advert() || toyota_camera_2aa() || toyota_prius_filter()) && ((msg->addr == 0x262U) || (!toyota_alt_cruise && (msg->addr == 0x1D3U)))) {
toyota_aol_observe(msg);
}
if (toyota_alt_cruise && msg_matches(msg, 0x365U, 0U) && (GET_LEN(msg) == 7U)) {
@@ -628,7 +632,7 @@ static safety_config toyota_init(uint16_t param) {
toyota_alt_brake = GET_FLAG(param, TOYOTA_PARAM_ALT_BRAKE);
toyota_stock_longitudinal = GET_FLAG(param, TOYOTA_PARAM_STOCK_LONGITUDINAL);
toyota_radar_diagnostics_allowed = !toyota_stock_longitudinal && !toyota_secoc && (param != 0x1049U) && (param != 0x1449U) && !toyota_eps8_sdsu && !toyota_legacy_sdsu && !toyota_legacy_advert_param(param) && !toyota_camera_2aa_param(param) && !toyota_prius_sdsu_param(param);
toyota_radar_diagnostics_allowed = !toyota_stock_longitudinal && !toyota_secoc && (param != 0x1049U) && (param != 0x1449U) && !toyota_eps8_sdsu && !toyota_legacy_sdsu && !toyota_legacy_advert_param(param) && !toyota_camera_2aa_param(param) && !toyota_prius_filter_param(param);
toyota_lta = !toyota_legacy_advert_param(param) && GET_FLAG(param, TOYOTA_PARAM_LTA);
toyota_dbc_eps_torque_factor = param & TOYOTA_EPS_FACTOR;
@@ -649,7 +653,7 @@ static safety_config toyota_init(uint16_t param) {
if (toyota_stock_longitudinal) {
SET_TX_MSGS(TOYOTA_TX_MSGS, ret);
} else {
if ((param == 0x1049U) || (param == 0x9049U) || (param == 0x1449U) || toyota_eps8_sdsu || toyota_legacy_sdsu || toyota_legacy_advert_param(param) || toyota_camera_2aa_param(param) || toyota_prius_sdsu_param(param)) {
if ((param == 0x1049U) || (param == 0x9049U) || (param == 0x1449U) || toyota_eps8_sdsu || toyota_legacy_sdsu || toyota_legacy_advert_param(param) || toyota_camera_2aa_param(param) || toyota_prius_filter_param(param)) {
SET_TX_MSGS(TOYOTA_FILTER_LONG_TX_MSGS_SET, ret);
} else {
SET_TX_MSGS(TOYOTA_LONG_TX_MSGS, ret);
@@ -688,8 +692,8 @@ static safety_config toyota_init(uint16_t param) {
// Restore the sixth row on every warm switch before specializing the new owner.
toyota_rx_checks[5].msg[0] = (CanMsgCheck){.addr = 0x365, .bus = 0U, .len = 7, .frequency = 10U,
.ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true};
if (toyota_legacy_advert_param(param) || toyota_camera_2aa_param(param)) {
toyota_rx_checks[4].msg[0] = (CanMsgCheck){.addr = 0x262, .bus = 0U, .len = toyota_camera_2aa_param(param) ? 8 : 5, .frequency = 25U,
if (toyota_legacy_advert_param(param) || toyota_camera_2aa_param(param) || toyota_prius_advert_param(param)) {
toyota_rx_checks[4].msg[0] = (CanMsgCheck){.addr = 0x262, .bus = 0U, .len = (toyota_camera_2aa_param(param) || toyota_prius_advert_param(param)) ? 8 : 5, .frequency = 25U,
.ignore_counter = true, .ignore_quality_flag = true};
if (!toyota_alt_cruise) {
toyota_rx_checks[5].msg[0] = (CanMsgCheck){.addr = 0x1D3, .bus = 0U, .len = 8, .frequency = 10U,
@@ -706,7 +710,7 @@ static safety_config toyota_init(uint16_t param) {
ret.disable_forwarding = false;
if (toyota_prius_sdsu_param(param)) {
ret.rx_checks_len = 7;
} else if ((toyota_legacy_sdsu && toyota_alt_cruise) || toyota_legacy_advert_param(param) || toyota_camera_2aa_param(param)) {
} else if ((toyota_legacy_sdsu && toyota_alt_cruise) || toyota_legacy_advert_param(param) || toyota_camera_2aa_param(param) || toyota_prius_advert_param(param)) {
ret.rx_checks_len = 6;
} else if (toyota_pedal || toyota_alt_cruise || toyota_sdsu_status) {
ret.rx_checks_len = 5;
@@ -715,7 +719,7 @@ static safety_config toyota_init(uint16_t param) {
}
}
if (((param & 0xC000U) != 0U) && !toyota_alt_cruise && !toyota_sdsu_status && !toyota_legacy_advert_param(param) && !toyota_camera_2aa_param(param)) {
if (((param & 0xC000U) != 0U) && !toyota_alt_cruise && !toyota_sdsu_status && !toyota_legacy_advert_param(param) && !toyota_camera_2aa_param(param) && !toyota_prius_advert_param(param)) {
ret.tx_msgs_len = 0;
}
if (((param & 0x2000U) != 0U) && !toyota_pedal) {
+22 -5
View File
@@ -8,6 +8,23 @@ static bool toyota_prius_sdsu(void) {
return toyota_prius_sdsu_param(current_safety_param);
}
static bool toyota_prius_advert_param(uint16_t param) {
return (param == 0xC042U) || (param == 0xC242U);
}
static bool toyota_prius_advert(void) {
return toyota_prius_advert_param(current_safety_param);
}
static bool toyota_prius_filter_param(uint16_t param) {
return toyota_prius_sdsu_param(param) || toyota_prius_advert_param(param);
}
static bool toyota_prius_filter(void) {
return toyota_prius_filter_param(current_safety_param);
}
// Disjoint original Prius owners share this sticky receiver identity latch.
static bool toyota_prius_sdsu_ambiguous = false;
static bool toyota_camera_2aa_param(uint16_t param) {
@@ -82,7 +99,7 @@ static uint8_t toyota_aol_permission(void) {
uint8_t permission = 0U;
if (fresh && toyota_aol_main && toyota_aol_drive && toyota_aol_doors_closed && toyota_aol_eps &&
(!toyota_camera_2aa() || !toyota_camera_2aa_ambiguous) &&
(!toyota_prius_sdsu() || !toyota_prius_sdsu_ambiguous)) {
(!toyota_prius_filter() || !toyota_prius_sdsu_ambiguous)) {
permission = request & 1U;
if (!toyota_aol_stock && controls_allowed && toyota_aol_belt && !gas_pressed && !brake_pressed &&
(!toyota_sdsu_status || toyota_sdsu_control)) {
@@ -102,7 +119,7 @@ static void toyota_aol_observe(const CANPacket_t *msg) {
expected_len = 5U;
} else {
}
if ((toyota_aol_enabled || toyota_legacy_advert() || toyota_camera_2aa() || toyota_prius_sdsu()) && (msg->bus == 0U) && (GET_LEN(msg) != expected_len)) {
if ((toyota_aol_enabled || toyota_legacy_advert() || toyota_camera_2aa() || toyota_prius_filter()) && (msg->bus == 0U) && (GET_LEN(msg) != expected_len)) {
unsigned int index = 4U;
if (msg->addr == main_addr) {
index = 0U;
@@ -117,7 +134,7 @@ static void toyota_aol_observe(const CANPacket_t *msg) {
if (index < 4U) {
toyota_aol_seen[index] = false;
}
} else if ((toyota_aol_enabled || toyota_legacy_advert() || toyota_camera_2aa() || toyota_prius_sdsu()) && (msg->bus == 0U) && (GET_LEN(msg) == expected_len)) {
} else if ((toyota_aol_enabled || toyota_legacy_advert() || toyota_camera_2aa() || toyota_prius_filter()) && (msg->bus == 0U) && (GET_LEN(msg) == expected_len)) {
unsigned int index = 4U;
if (msg->addr == main_addr) {
index = 0U;
@@ -166,13 +183,13 @@ static void toyota_aol_configure(uint16_t param) {
toyota_pedal = (param == 0x2049U) || (param == 0x2249U);
toyota_aol_rav4h_filter = (param == 0x9049U) || (param == 0x9249U);
toyota_sdsu_status = toyota_aol_rav4h_filter || toyota_eps8_sdsu || toyota_legacy_sdsu || toyota_prius_sdsu_param(param);
toyota_aol_stock = (param == 585U) || (param == 4681U) || (param == 0x9249U) || (param == 0x2249U) || (param == 0x424DU) || (param == 0x649U) || (param == 0x1649U) || (param == 0x8249U) || (param == 0x8649U) || ((toyota_legacy_sdsu || toyota_legacy_advert_param(param) || toyota_camera_2aa_param(param) || toyota_prius_sdsu_param(param)) && ((param & 0x200U) != 0U));
toyota_aol_stock = (param == 585U) || (param == 4681U) || (param == 0x9249U) || (param == 0x2249U) || (param == 0x424DU) || (param == 0x649U) || (param == 0x1649U) || (param == 0x8249U) || (param == 0x8649U) || ((toyota_legacy_sdsu || toyota_legacy_advert_param(param) || toyota_camera_2aa_param(param) || toyota_prius_filter_param(param)) && ((param & 0x200U) != 0U));
#ifdef ALLOW_DEBUG
toyota_aol_secoc = (param == 2121U);
#else
toyota_aol_secoc = false;
#endif
toyota_aol_enabled = ((toyota_eps8_sdsu || toyota_legacy_sdsu || toyota_legacy_advert_param(param) || toyota_camera_2aa_param(param) || toyota_prius_sdsu_param(param)) && (alternative_experience == 32)) || (toyota_aol_lta && (alternative_experience == 32)) || (toyota_aol_stock && (alternative_experience == 32)) ||
toyota_aol_enabled = ((toyota_eps8_sdsu || toyota_legacy_sdsu || toyota_legacy_advert_param(param) || toyota_camera_2aa_param(param) || toyota_prius_filter_param(param)) && (alternative_experience == 32)) || (toyota_aol_lta && (alternative_experience == 32)) || (toyota_aol_stock && (alternative_experience == 32)) ||
#ifdef ALLOW_DEBUG
(toyota_aol_secoc && (alternative_experience == 32)) ||
#endif
@@ -0,0 +1,150 @@
"""Original Prius four-byte identity never provides a status or source lease."""
import pytest
from opendbc.car import structs
from opendbc.car.toyota import prius_advert_filter as owner, prius_sdsu
from opendbc.safety.tests.libsafety import libsafety_py
from opendbc.safety.tests.test_toyota_prius_sdsu import Physical
from openpilot.starpilot.car.toyota.tests import test_rav4h_caller_native as helpers
source_matched_native = helpers.source_matched_native
class AdvertPhysical(Physical):
def tick(self, cruise=True, *, status=0, malformed=False, mirror=False, source=None, failure=None, conflict=None, ad=True):
assert status == 0 and not malformed and not mirror, 'Advertisement profiles have no SDSU status input'
self.now += 10_000
self.native.set_timer(self.now)
frames = [frame for frame in helpers.physical(self.packer, cruise=cruise) if frame[0] != 0x2FF]
if ad:
# Nibble7 is not a status or AEB source; a mirror cannot replace MAIN/EPS.
frames.extend(((0x2FF, bytes.fromhex('00007409'), 0), (0x2FF, bytes.fromhex('80007409'), 2)))
if conflict is not None:
frames.append(conflict)
for address, raw, bus in frames:
expected = True
if address == source and bus == 0:
if failure == 'missing':
continue
if failure == 'checksum':
raw = raw[:-1] + bytes([raw[-1] ^ 1])
expected = False
elif failure == 'length':
raw = raw[:-1]
elif failure == 'mirror':
bus = 2
assert bool(self.native.safety_rx_hook(libsafety_py.make_CANPacket(address, bus, raw))) == expected
self.native.safety_tick()
self.native.aol_set_host_request(3)
def active(case, stock, experience):
assert case.native.safety_config_valid() and case.native.get_controls_allowed()
assert case.native.aol_get_permission_mask() == ((1 if stock else 3) if experience else 0)
assert case.tx('STEERING_LKA', {'STEER_REQUEST': 1, 'STEER_TORQUE_CMD': 3})
assert case.tx('ACC_CONTROL', {'ACCEL_CMD': 0.1}) == (not stock)
def refused(case):
assert case.native.aol_get_permission_mask() == 0
assert not case.tx('STEERING_LKA', {'STEER_REQUEST': 1, 'STEER_TORQUE_CMD': 3})
assert not case.tx('ACC_CONTROL', {'ACCEL_CMD': 0.1})
assert case.tx('ACC_CONTROL', {'ACCEL_CMD': 0, 'CANCEL_REQ': 1})
@pytest.mark.parametrize('stock', (False, True))
@pytest.mark.parametrize('experience', (0, 32))
def test_original_prius_advert_six_independent_rx_and_limits(stock, experience):
case = AdvertPhysical(owner.STOCK_WORD if stock else owner.LONG_WORD, experience)
try:
cfg = case.native.current_safety_config
assert cfg.rx_checks_len == 6
checks = {cfg.rx_checks[i].msg[0].addr: cfg.rx_checks[i].msg[0] for i in range(6)}
assert set(checks) == {0xAA, 0x260, 0x1D2, 0x226, 0x262, 0x1D3}
assert all(checks[a].len == 8 and not checks[a].ignore_checksum for a in (0x262, 0x1D3))
assert not case.tx('STEERING_LKA', {'STEER_REQUEST': 1, 'STEER_TORQUE_CMD': 3})
case.earn()
active(case, stock, experience)
for _ in range(50):
case.tick(ad=False)
active(case, stock, experience)
for accel in (-3.501, 2.001):
assert not case.tx('ACC_CONTROL', {'ACCEL_CMD': accel})
assert not case.tx('STEERING_LTA', {'STEER_REQUEST': 1})
assert not case.native.safety_tx_hook(libsafety_py.make_CANPacket(0x750, 0, bytes.fromhex('0f02280100000000')))
assert case.native.safety_fwd_hook(2, 0x343) == 0
finally:
case.close()
@pytest.mark.parametrize('stock', (False, True))
@pytest.mark.parametrize('experience', (0, 32))
@pytest.mark.parametrize('source', (0x1D3, 0x262))
@pytest.mark.parametrize('failure', ('checksum', 'length', 'missing', 'mirror'))
def test_original_prius_advert_main_eps_refusal_and_recovery(stock, experience, source, failure):
case = AdvertPhysical(owner.STOCK_WORD if stock else owner.LONG_WORD, experience)
try:
case.earn()
cfg = case.native.current_safety_config
index = 5 if source == 0x1D3 else 4
prior = int(cfg.rx_checks[index].status.last_timestamp)
for _ in range(41):
case.tick(source=source, failure=failure)
if failure == 'checksum':
assert int(cfg.rx_checks[index].status.last_timestamp) == case.now
assert not cfg.rx_checks[index].status.valid_checksum
else:
assert int(cfg.rx_checks[index].status.last_timestamp) == prior
refused(case)
case.earn()
active(case, stock, experience)
finally:
case.close()
@pytest.mark.parametrize('stock', (False, True))
@pytest.mark.parametrize('experience', (0, 32))
@pytest.mark.parametrize('address,bus,length', ((0x201, 0, 6), (0x2AA, 0, 8), (0x2FF, 1, 4), (0x2FF, 0, 8), (0x2FF, 2, 8), (0x2FF, 0, 7)))
def test_original_prius_advert_sticky_conflict_until_reconfiguration(stock, experience, address, bus, length):
word = owner.STOCK_WORD if stock else owner.LONG_WORD
case = AdvertPhysical(word, experience)
try:
case.earn()
case.tick(conflict=(address, bytes(length), bus))
case.tick(source=0x262, failure='checksum')
case.earn()
assert case.native.safety_config_valid()
refused(case)
assert case.native.set_safety_hooks(structs.CarParams.SafetyModel.toyota, word) == 0
case.native.set_aol_test_heartbeat(True)
cfg = case.native.current_safety_config
assert cfg.rx_checks_len == 6
assert all(cfg.rx_checks[i].status.last_timestamp == 0 for i in range(6))
case.earn()
active(case, stock, experience)
finally:
case.close()
@pytest.mark.parametrize('stock', (False, True))
@pytest.mark.parametrize('experience', (0, 32))
def test_original_prius_advert_warm_switch_keeps_separate_status_owner(stock, experience):
case = AdvertPhysical(owner.STOCK_WORD if stock else owner.LONG_WORD, experience)
try:
case.earn()
assert case.native.set_safety_hooks(structs.CarParams.SafetyModel.toyota, prius_sdsu.STOCK_WORD if stock else prius_sdsu.LONG_WORD) == 0
case.native.set_aol_test_heartbeat(True)
assert case.native.current_safety_config.rx_checks_len == 7
assert not case.tx('STEERING_LKA', {'STEER_REQUEST': 1, 'STEER_TORQUE_CMD': 3})
# The old owner earns status from genuine eight-byte SDSU traffic.
for cruise in (False, True):
for _ in range(25):
Physical.tick(case, cruise)
active(case, stock, experience)
assert case.native.set_safety_hooks(structs.CarParams.SafetyModel.toyota, owner.STOCK_WORD if stock else owner.LONG_WORD) == 0
case.native.set_aol_test_heartbeat(True)
assert case.native.current_safety_config.rx_checks_len == 6
case.earn()
active(case, stock, experience)
finally:
case.close()
@@ -179,7 +179,7 @@ int main() {
}
for (uint32_t word = 0U; word <= 65535U; ++word) {
const bool expected = (word == 0x8042U) || (word == 0x8242U) || (word == 0xC049U) || (word == 0xC249U) || (word == 73U) || (word == 585U) || (word == 4169U) || (word == 4681U) || (word == 2121U) || (word == 0x9049U) || (word == 0x9249U) || (word == 0x2049U) || (word == 0x2249U) || (word == 0x424DU) || (word == 0x449U) || (word == 0x649U) || (word == 0x1449U) || (word == 0x1649U) || (word == 0x8049U) || (word == 0x8249U) || (word == 0x8449U) || (word == 0x8649U) || (word == 0x9158U) || (word == 0x9358U) || (word == 0x9149U) || (word == 0x9349U) || (word == 0x9164U) || (word == 0x9364U) || (word == 0xd149U) || (word == 0xd349U) || (word == 0xd04dU) || (word == 0xd24dU) || (word == 0x9449U) || (word == 0x9549U) || (word == 0x9558U) || (word == 0x9564U) || (word == 0x9649U) || (word == 0x9749U) || (word == 0x9758U) || (word == 0x9764U) || (word == 0xD44DU) || (word == 0xD549U) || (word == 0xD64DU) || (word == 0xD749U);
const bool expected = (word == 0xC042U) || (word == 0xC242U) || (word == 0x8042U) || (word == 0x8242U) || (word == 0xC049U) || (word == 0xC249U) || (word == 73U) || (word == 585U) || (word == 4169U) || (word == 4681U) || (word == 2121U) || (word == 0x9049U) || (word == 0x9249U) || (word == 0x2049U) || (word == 0x2249U) || (word == 0x424DU) || (word == 0x449U) || (word == 0x649U) || (word == 0x1449U) || (word == 0x1649U) || (word == 0x8049U) || (word == 0x8249U) || (word == 0x8449U) || (word == 0x8649U) || (word == 0x9158U) || (word == 0x9358U) || (word == 0x9149U) || (word == 0x9349U) || (word == 0x9164U) || (word == 0x9364U) || (word == 0xd149U) || (word == 0xd349U) || (word == 0xd04dU) || (word == 0xd24dU) || (word == 0x9449U) || (word == 0x9549U) || (word == 0x9558U) || (word == 0x9564U) || (word == 0x9649U) || (word == 0x9749U) || (word == 0x9758U) || (word == 0x9764U) || (word == 0xD44DU) || (word == 0xD549U) || (word == 0xD64DU) || (word == 0xD749U);
auto toyota = status();
toyota.safety_mode = 2U;
toyota.safety_param = static_cast<uint16_t>(word);
+1 -1
View File
@@ -6,7 +6,7 @@ from openpilot.starpilot.aol.policy import AolVehiclePolicy
from openpilot.starpilot.aol.intent import AolCardIntent, AOL_TOGGLE
AOL_WORDS = frozenset((
0x8042, 0x8242,
0x8042, 0x8242, 0xC042, 0xC242,
0xC049, 0xC249,
0x9449, 0x9549, 0x9558, 0x9564, 0x9649, 0x9749, 0x9758, 0x9764, 0xD44D, 0xD549, 0xD64D, 0xD749,
73, 585, 4169, 4681, 2121, 0x9049, 0x9249,
+1 -1
View File
@@ -3,7 +3,7 @@
#include "selfdrive/pandad/aol_protocol.h"
inline bool toyota_aol_param(uint16_t param) {
return (param == 0x8042U || param == 0x8242U || param == 0xC049U) || (param == 0xC249U) || (param == 73U) || (param == 585U) || (param == 4169U) || (param == 4681U) || (param == 2121U) || (param == 0x9049U) || (param == 0x9249U) || (param == 0x2049U) || (param == 0x2249U) || (param == 0x424DU) || (param == 0x449U) || (param == 0x649U) || (param == 0x1449U) || (param == 0x1649U) || (param == 0x8049U) || (param == 0x8249U) || (param == 0x8449U) || (param == 0x8649U) || (param == 0x9158U) || (param == 0x9358U) || (param == 0x9149U) || (param == 0x9349U) || (param == 0x9164U) || (param == 0x9364U) || (param == 0xd149U) || (param == 0xd349U) || (param == 0xd04dU) || (param == 0xd24dU) || (param == 0x9449U) || (param == 0x9549U) || (param == 0x9558U) || (param == 0x9564U) || (param == 0x9649U) || (param == 0x9749U) || (param == 0x9758U) || (param == 0x9764U) || (param == 0xD44DU) || (param == 0xD549U) || (param == 0xD64DU) || (param == 0xD749U);
return (param == 0xC042U || param == 0xC242U || param == 0x8042U || param == 0x8242U || param == 0xC049U) || (param == 0xC249U) || (param == 73U) || (param == 585U) || (param == 4169U) || (param == 4681U) || (param == 2121U) || (param == 0x9049U) || (param == 0x9249U) || (param == 0x2049U) || (param == 0x2249U) || (param == 0x424DU) || (param == 0x449U) || (param == 0x649U) || (param == 0x1449U) || (param == 0x1649U) || (param == 0x8049U) || (param == 0x8249U) || (param == 0x8449U) || (param == 0x8649U) || (param == 0x9158U) || (param == 0x9358U) || (param == 0x9149U) || (param == 0x9349U) || (param == 0x9164U) || (param == 0x9364U) || (param == 0xd149U) || (param == 0xd349U) || (param == 0xd04dU) || (param == 0xd24dU) || (param == 0x9449U) || (param == 0x9549U) || (param == 0x9558U) || (param == 0x9564U) || (param == 0x9649U) || (param == 0x9749U) || (param == 0x9758U) || (param == 0x9764U) || (param == 0xD44DU) || (param == 0xD549U) || (param == 0xD64DU) || (param == 0xD749U);
}
inline constexpr AolSafetyProfile TOYOTA_AOL_PROFILE{2U, toyota_aol_param, true};
@@ -0,0 +1,199 @@
"""Draft actual full caller catalog for original Prius four-byte advertisements."""
from dataclasses import replace
import pytest
from opendbc.car import Bus
from opendbc.car.toyota import prius_advert_filter as owner
from opendbc.car.toyota.values import CAR
from openpilot.starpilot.aol.wire import decode_safety, encode_safety
from openpilot.starpilot.car.toyota.tests import test_rav4h_caller_native as helpers
from openpilot.starpilot.car.toyota.tests.test_prius_sdsu_filter import original_firmware, ORIGINAL
from openpilot.starpilot.car.toyota.tests.test_toyota_cpp_transport import host_heartbeat_bridge as host_heartbeat_bridge
source_matched_native = helpers.source_matched_native
daemon_gc_scope = helpers.daemon_gc_scope
def run(version, stock, bridge, scenario='nominal', *, aol=True, fault=None, mirror=False):
tick = 0
def physical(packer, **fields):
nonlocal tick
tick += 1
if scenario == 'physical':
if 410 <= tick < 415:
fields['gear'] = 32
if 475 <= tick < 480:
fields.update(main=False, cruise=False)
if scenario == 'identity_refusal':
fields.update(cruise=tick >= 150 and not 330 <= tick < 360, main=not 340 <= tick < 350)
if fault in ('eps-temporary', 'eps-permanent') and 260 <= tick < 311:
fields['eps'] = 9 if fault == 'eps-temporary' else 3
if fault == 'configured-brake' and 260 <= tick < 311:
fields.update(brake=True, cruise=False, stopped=False)
frames = helpers.physical(packer, **fields)
if scenario == 'physical' and 470 <= tick < 475:
frames = [f for f in frames if f[0] != 0x620]
frames.append(packer.make_can_msg('BODY_CONTROL_STATE', 0, {'DOOR_OPEN_FL': 1}))
frames = [f for f in frames if f[0] != 0x2FF]
frames += [(0x2FF, bytes.fromhex('80007409' if 220 <= tick < 240 else '00007409'), 0)]
if mirror:
frames.append((0x2FF, bytes.fromhex('80007409'), 2))
if fault == 'aeb' and 260 <= tick < 311:
frames = [f for f in frames if f[0] != 0x283]
frames.append(packer.make_can_msg('PRE_COLLISION', 0, {'PRECOLLISION_ACTIVE': 1, 'FORCE': -20}))
if fault is not None and fault.startswith('identity-') and tick == 260:
markers = {
'identity-advert': (0x2AA, bytes(8), 0),
'identity-pedal': (0x201, bytes(6), 0),
'identity-bus': (0x2FF, bytes(8), 1),
'identity-status-advert': (0x2FF, bytes(8), 0),
'identity-mirror-advert': (0x2FF, bytes(8), 2),
'identity-length': (0x2FF, bytes(7), 0),
}
frames.append(markers[fault])
if fault is not None and fault.startswith('identity-') and tick == 275:
# A genuine invalid-source reset must not erase contradictory identity.
frames = [frame for frame in frames if frame[0] != 0x262]
a, data, bus = packer.make_can_msg('EPS_STATUS', 0, {'LKA_STATE': 1, 'LTA_STATE': 1})
frames.append((a, data[:-1] + bytes([data[-1] ^ 1]), bus))
if fault is not None and fault.split('-')[0] in ('main', 'eps') and fault not in ('eps-temporary', 'eps-permanent') and 260 <= tick < 311:
source, kind = fault.split('-')
address = 0x1D3 if source == 'main' else 0x262
kept = []
for a, data, bus in frames:
if a == address and bus == 0:
if kind == 'drop':
continue
if kind == 'length':
data = data[:-1]
elif kind == 'checksum':
data = data[:-1] + bytes([data[-1] ^ 1])
else:
assert kind == 'bus'
bus = 2
kept.append((a, data, bus))
frames = kept
return frames
def observe(tick, host, cs, cc, safety, native):
del native
parser = host.CI.can_parsers[Bus.pt]
assert isinstance(parser, owner.PriusAdvertCANParser)
assert 0x2FF not in parser.message_states and 'SDSU' not in parser.vl
if tick == 220:
assert host.CI.CS.distance_button == parser.advert_distance_button == 1
if tick == 240:
assert host.CI.CS.distance_button == parser.advert_distance_button == 0, 'Mirror button cannot replace primary release'
if 200 <= tick < 250:
assert not cs.stockAeb, 'Advertisement nibble7 is not SDSU STATE/AEB'
assert host.CI.CS.eps_torque_scale == 0.66
assert abs(host.CP.longitudinalActuatorDelay - 0.15) < 1e-6
assert host.CP.safetyConfigs[0].safetyParam == (owner.STOCK_WORD if stock else owner.LONG_WORD)
assert safety.safety_fwd_hook(2, 0x343) == 0 and safety.safety_fwd_hook(2, 0x4CB) == 0
if scenario == 'physical' and (410 <= tick < 430 or 470 <= tick < 475):
assert host.CI.CC.last_torque == 0
if not stock:
assert host.CI.CC.standstill_req
if fault == 'aeb' and 270 <= tick < 300:
assert cs.stockAeb and not cc.latActive and not cc.longActive
if fault is not None and 301 <= tick < 308:
assert not cc.latActive and not cc.longActive and safety.aol_get_permission_mask() == 0
if scenario == 'identity_refusal' and tick >= 270:
assert parser.hardware_ambiguous
assert not cc.latActive and not cc.longActive and safety.aol_get_permission_mask() == 0
if tick >= 390:
assert safety.safety_config_valid(), 'all real required sources recovered, but identity must stay refused'
def rx_expected(address, data, bus, default, cp):
if fault is not None and fault.startswith('identity-') and tick == 275 and address == 0x262 and bus == 0:
from opendbc.car.toyota.toyotacan import toyota_checksum
assert owner.configured(cp) and len(data) == 8 and data[-1] != toyota_checksum(address, None, bytearray(data))
return False
if fault == 'eps-checksum' and address == 0x262 and bus == 0 and 260 <= tick < 311:
assert owner.configured(cp) and len(data) == 8
return False
if fault == 'main-checksum' and address == 0x1D3 and bus == 0 and 260 <= tick < 311:
assert owner.configured(cp) and len(data) == 8
return False
return default
original_native = helpers.native_events
def native_events(*args, **kwargs):
messages = original_native(*args, **kwargs)
if fault in ('native-identity', 'native-session') and 260 <= tick < 311:
original = next(event for event in messages if event.which() == 'aolSafetyWire')
reader = original.as_reader() if hasattr(original, 'as_reader') else original
wire = reader.as_builder()
state = decode_safety(bytes(wire.aolSafetyWire))
assert state is not None
# USB producer boundary: change identity/session only, preserving all actual native bits.
wrong = replace(state, safetyParam=state.safetyParam ^ 1) if fault == 'native-identity' else replace(state, axisSessionId='f' * 32)
wire.aolSafetyWire = encode_safety(wrong)
messages = [wire.as_reader() if event.which() == 'aolSafetyWire' else event for event in messages]
return messages
with pytest.MonkeyPatch.context() as monkeypatch:
if fault in ('native-identity', 'native-session'):
monkeypatch.setattr(helpers, 'native_events', native_events)
helpers.run_case(
stock,
scenario,
candidate=CAR.TOYOTA_PRIUS,
firmware_override=original_firmware(version),
bridge=bridge,
input_clock=True,
aol=aol,
sdsu_words=(owner.LONG_WORD, owner.STOCK_WORD),
physical_frames=physical,
fingerprint_frames=[(0x2FF, bytes(4), 0)] + ([(0x2FF, bytes(4), 2)] if mirror else []),
on_sample=observe,
native_rx_expected=rx_expected,
brake_stop_check_start=355,
brake_pause_mps=25.0 if fault == 'configured-brake' else None,
)
@pytest.mark.parametrize('version', ORIGINAL)
@pytest.mark.parametrize('stock', (False, True))
def test_original_prius_advert_all_known_firmware(version, stock, host_heartbeat_bridge):
run(version, stock, host_heartbeat_bridge, mirror=True)
@pytest.mark.parametrize('version', (ORIGINAL[0], next(v for v in ORIGINAL if v.startswith(b'8965B47060'))))
@pytest.mark.parametrize('stock', (False, True))
@pytest.mark.parametrize('scenario', ('physical', 'off', 'cold_low', 'ordinary_aol_off'))
def test_original_prius_advert_physical_and_preferences(version, stock, scenario, host_heartbeat_bridge):
run(version, stock, host_heartbeat_bridge, 'nominal' if scenario == 'ordinary_aol_off' else scenario, aol=scenario != 'ordinary_aol_off', mirror=True)
@pytest.mark.parametrize('stock', (False, True))
@pytest.mark.parametrize(
'fault',
(
'main-drop',
'main-length',
'main-checksum',
'main-bus',
'eps-drop',
'eps-length',
'eps-checksum',
'eps-bus',
'eps-temporary',
'eps-permanent',
'native-identity',
'native-session',
'identity-advert',
'identity-pedal',
'identity-bus',
'identity-status-advert',
'identity-mirror-advert',
'identity-length',
'configured-brake',
'aeb',
),
)
def test_original_prius_advert_actual_source_refusals(stock, fault, host_heartbeat_bridge):
run(ORIGINAL[0], stock, host_heartbeat_bridge, 'identity_refusal' if fault.startswith('identity-') else 'main_lease', fault=fault, mirror=True)
@@ -0,0 +1,32 @@
"""Prius four-byte regressions remain mandatory in serial and native CI."""
import ast
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[5]
CALLER = 'openpilot/starpilot/car/toyota/tests/test_prius_advert_callers.py'
NATIVE = 'opendbc_repo/opendbc/safety/tests/test_toyota_prius_advert.py'
FACTORY = 'openpilot/starpilot/car/toyota/tests/test_prius_advert_filter.py'
def constant(name):
module = ast.parse((ROOT / 'tools/ci/run_vehicle_tests.py').read_text())
return next(ast.literal_eval(node.value) for node in module.body if isinstance(node, ast.Assign)
and any(isinstance(target, ast.Name) and target.id == name for target in node.targets))
def test_prius_advert_mandatory_native_and_joined_targets():
lifecycle = constant('AOL_LIFECYCLE_TARGETS')
assert all(lifecycle.count(path) == 1 for path in (CALLER, NATIVE, FACTORY))
assert constant('RELEASE_TARGETS').count(NATIVE) == 1
assert (ROOT / NATIVE).name.startswith('test_')
def test_prius_advert_serial_group_bounds_and_timing():
custom = json.loads((ROOT / 'tools/ci/custom_pytest_workloads.json').read_text())
lifecycle = json.loads((ROOT / 'tools/ci/lifecycle_workloads.json').read_text())
assert custom['independent_functions'][CALLER]['seconds_per_case'] == 4.0
assert lifecycle['function_groups'][CALLER]['estimated_seconds_per_case'] == 4.0
assert lifecycle['function_groups'][CALLER]['independent_parameters']
assert custom['independent_max_cases'] <= 60
assert custom['independent_max_seconds'] <= 240
@@ -0,0 +1,90 @@
"""Four-byte Prius admission is independent of eight-byte status and EPS73."""
import pytest
from opendbc.can import CANPacker
from opendbc.car import Bus, gen_empty_fingerprint, structs
from opendbc.car.toyota import prius_advert_filter as owner
from opendbc.car.toyota.interface import CarInterface
from opendbc.car.toyota.values import CAR, DBC
from openpilot.starpilot.car.toyota.tests.test_prius_sdsu_filter import original_firmware, ORIGINAL
@pytest.mark.parametrize(
'bad',
(
'unknown_eps',
'unknown_camera',
'missing_eps',
'missing_camera',
'logging_eps',
'logging_camera',
'wrong_brand',
'wrong_fw_bus',
'pedal',
'twoaa',
'bus1',
'mixed_mirror8',
'bad_primary7',
'bad_eps5',
'bad_main7',
'retrofit47070',
),
)
def test_original_prius_advert_refuses_unqualified_factory_sources(bad):
fw = original_firmware(ORIGINAL[0])
fp = gen_empty_fingerprint()
fp[0][0x2FF] = 4
if bad.startswith(('unknown_', 'missing_', 'logging_')):
ecu = structs.CarParams.Ecu.eps if bad.endswith('eps') else structs.CarParams.Ecu.fwdCamera
if bad.startswith('missing'):
fw = [item for item in fw if item.ecu != ecu]
else:
for item in fw:
if item.ecu == ecu:
if bad.startswith('unknown'):
item.fwVersion = b'unknown'
else:
item.logging = True
elif bad in ('wrong_brand', 'wrong_fw_bus'):
eps = next(item for item in fw if item.ecu == structs.CarParams.Ecu.eps)
if bad == 'wrong_brand':
eps.brand = 'other'
else:
eps.bus = 1
elif bad == 'pedal':
fp[0][0x201] = 6
elif bad == 'twoaa':
fp[0][0x2AA] = 8
elif bad == 'bus1':
fp[1][0x2FF] = 4
elif bad == 'mixed_mirror8':
fp[2][0x2FF] = 8
elif bad == 'bad_primary7':
fp[0][0x2FF] = 7
elif bad == 'bad_eps5':
fp[0][0x262] = 5
elif bad == 'bad_main7':
fp[0][0x1D3] = 7
else:
fw = original_firmware(owner.RETROFIT_EPS)
assert not owner.hardware_supported(CAR.TOYOTA_PRIUS, fp, fw)
cp = CarInterface.get_params(CAR.TOYOTA_PRIUS, fp, fw, True, False, False)
assert not owner.configured(cp) and cp.safetyConfigs[0].safetyParam not in owner.WORDS
@pytest.mark.parametrize('address,bus,length', ((0x201, 0, 6), (0x2AA, 0, 8), (0x2FF, 1, 4), (0x2FF, 0, 8), (0x2FF, 2, 8), (0x2FF, 0, 7)))
def test_original_prius_advert_parser_conflict_survives_source_recovery(address, bus, length):
packer = CANPacker(DBC[CAR.TOYOTA_PRIUS][Bus.pt])
parser = owner.PriusAdvertCANParser(DBC[CAR.TOYOTA_PRIUS][Bus.pt], [('PCM_CRUISE_2', 10), ('EPS_STATUS', 25)], 0)
main = packer.make_can_msg('PCM_CRUISE_2', 0, {'MAIN_ON': 1, 'SET_SPEED': 90})
eps = packer.make_can_msg('EPS_STATUS', 0, {'LKA_STATE': 1})
parser.update([(2_000_000_000, [main, eps, (0x2FF, bytes(4), 0), (0x2FF, bytes(4), 2)])])
assert parser.main_valid and parser.eps_valid and not parser.hardware_ambiguous
parser.update([(2_010_000_000, [(address, bytes(length), bus)])])
assert parser.hardware_ambiguous
invalid = (eps[0], eps[1][:-1] + bytes([eps[1][-1] ^ 1]), 0)
parser.update([(2_020_000_000, [invalid])])
assert parser.hardware_ambiguous and not parser.eps_valid
parser.update([(2_030_000_000, [main, eps, (0x2FF, bytes(4), 0)])])
assert parser.hardware_ambiguous and parser.main_valid and parser.eps_valid
assert 0x2FF not in parser.message_states and 'SDSU' not in parser.vl
@@ -4,7 +4,7 @@ import json
from pathlib import Path
import pytest
from opendbc.car import gen_empty_fingerprint, structs
from opendbc.car.toyota import prius_sdsu, prius_longitudinal
from opendbc.car.toyota import prius_sdsu, prius_longitudinal, prius_advert_filter
from opendbc.car.toyota.carstate import CarState
from opendbc.car.toyota.interface import CarInterface
from opendbc.car.toyota.values import CAR, ToyotaFlags
@@ -28,7 +28,7 @@ def factory(version=ORIGINAL[0], *, hardware='2FF8', alpha=False, mirror=False):
if hardware != 'none':
fp[0][0x2FF] = int(hardware[-1])
if mirror:
fp[2][0x2FF] = 8
fp[2][0x2FF] = int(hardware[-1])
return CarInterface.get_params(CAR.TOYOTA_PRIUS, fp, original_firmware(version), alpha, False, False)
@@ -39,19 +39,20 @@ def test_full_factory_cp_and_numeric_tunes(label):
alpha = alpha == 'alpha=True'
cp = factory(version, hardware=hardware, alpha=alpha)
expected = deepcopy(BASE[label])
if version in ORIGINAL and hardware == '2FF8':
owner = prius_sdsu if hardware == '2FF8' else prius_advert_filter
if version in ORIGINAL and hardware in ('2FF4', '2FF8'):
expected['flags'] |= int(ToyotaFlags.HYBRID | ToyotaFlags.LONG_FILTER | ToyotaFlags.RAISED_ACCEL_LIMIT)
expected['openpilotLongitudinalControl'] = True
expected['safetyConfigs'][0]['safetyParam'] = prius_sdsu.LONG_WORD
assert prius_sdsu.configured(cp) and prius_longitudinal.enabled(cp)
assert factory(version, alpha=alpha, mirror=True).to_dict() == cp.to_dict()
expected['safetyConfigs'][0]['safetyParam'] = owner.LONG_WORD
assert owner.configured(cp) and prius_longitudinal.enabled(cp)
assert factory(version, hardware=hardware, alpha=alpha, mirror=True).to_dict() == cp.to_dict()
assert CarState(cp).eps_torque_scale == .66
actual = cp.to_dict()
assert actual == expected
assert prius_longitudinal.prepare_stock(cp)
expected['openpilotLongitudinalControl'] = False
expected['safetyConfigs'][0]['safetyParam'] = prius_sdsu.STOCK_WORD
assert cp.to_dict() == expected and prius_sdsu.configured(cp)
expected['safetyConfigs'][0]['safetyParam'] = owner.STOCK_WORD
assert cp.to_dict() == expected and owner.configured(cp)
assert CarState(cp).eps_torque_scale == .66
else:
assert not prius_sdsu.configured(cp)
+1 -1
View File
@@ -1 +1 @@
{"protected_inventory_sha256": "fb0f5469122a7bb2cf29f71b69a3d32709e54a94cdcb04026548828be6185226", "schema_version": 1}
{"protected_inventory_sha256": "01e53e88a4f6bfbe1d3b7f4adca886a6636e20b3bf058cf2d1b0e4c2df362902", "schema_version": 1}
+1
View File
@@ -19,6 +19,7 @@ HEAVY_MODULES = {
'openpilot/starpilot/car/toyota/tests/test_camera_2aa_callers.py',
'openpilot/starpilot/car/toyota/tests/test_advert_filter_callers.py',
'openpilot/starpilot/car/toyota/tests/test_prius_sdsu_callers.py',
'openpilot/starpilot/car/toyota/tests/test_prius_advert_callers.py',
'openpilot/starpilot/car/toyota/tests/test_legacy_sdsu_callers.py',
'openpilot/starpilot/car/toyota/tests/test_2aa_forwarded_copy_callers.py',
}
+4
View File
@@ -150,6 +150,10 @@
"basis": "6d hosted shard8: 205/240 cases completed in587.445481209 phase seconds before whole-shard900s cutoff; Fresh per-case caller/native state and cleanup source-verified; full in-case timeline remains unchanged.",
"seconds_per_case": 4.0
},
"openpilot/starpilot/car/toyota/tests/test_prius_advert_callers.py": {
"basis": "Scheduling estimate4s/case; all132 mirror caller cases actually passed across eight bounded debug/release phases; worst20case phase55.81s. Fresh per-case Params/rawCAN/native configuration/C++ reset and cleanup reviewed; no shared outcomes; full in-case timelines retained.",
"seconds_per_case": 4.0
},
"openpilot/starpilot/car/toyota/tests/test_prius_sdsu_callers.py": {
"basis": "90 actual DEBUG/RELEASE caller cases passed in four bounded phases (186.885s combined wall time including startup). Scheduling estimate 3s/case; fresh per-case Params, raw CAN, native configuration and C++ host reset; no shared outcome dependence.",
"seconds_per_case": 3.0
+2 -1
View File
@@ -93,7 +93,8 @@ def workload_policy():
if policy['version'] != 1 or set(policy['function_groups']) != {
'openpilot/starpilot/car/toyota/tests/test_legacy_advert_callers.py',
'openpilot/starpilot/car/toyota/tests/test_camera_2aa_callers.py',
'openpilot/starpilot/car/toyota/tests/test_prius_sdsu_callers.py'}:
'openpilot/starpilot/car/toyota/tests/test_prius_sdsu_callers.py',
'openpilot/starpilot/car/toyota/tests/test_prius_advert_callers.py'}:
raise ValueError('Unreviewed lifecycle function grouping policy')
if type(policy['independent_function_max_cases']) is not int or not 1 <= policy['independent_function_max_cases'] <= 60:
raise ValueError('Invalid independent function group size')
+5
View File
@@ -10,6 +10,11 @@
"estimated_seconds_per_case": 4.0,
"independent_parameters": true
},
"openpilot/starpilot/car/toyota/tests/test_prius_advert_callers.py": {
"basis": "Scheduling estimate4s/case; all132 mirror caller cases actually passed across eight bounded debug/release phases; worst20case phase55.81s. Fresh per-case Params/rawCAN/native configuration/C++ reset and cleanup reviewed; no shared outcomes; full in-case timelines retained.",
"estimated_seconds_per_case": 4.0,
"independent_parameters": true
},
"openpilot/starpilot/car/toyota/tests/test_prius_sdsu_callers.py": {
"basis": "90 actual DEBUG/RELEASE caller cases passed in four bounded phases (186.885s combined wall time including startup). Scheduling estimate 3s/case; fresh per-case Params, raw CAN, native configuration and C++ host reset; no shared outcome dependence.",
"estimated_seconds_per_case": 3.0,
+5
View File
@@ -100,6 +100,10 @@ AOL_LIFECYCLE_TARGETS = (
"openpilot/starpilot/car/toyota/tests/test_prius_sdsu_filter.py",
"openpilot/starpilot/car/toyota/tests/test_prius_sdsu_callers.py",
"opendbc_repo/opendbc/safety/tests/test_toyota_prius_sdsu.py",
"openpilot/starpilot/car/toyota/tests/test_prius_advert_filter.py",
"openpilot/starpilot/car/toyota/tests/test_prius_advert_callers.py",
"opendbc_repo/opendbc/safety/tests/test_toyota_prius_advert.py",
"openpilot/starpilot/car/toyota/tests/test_prius_advert_ci.py",
"openpilot/starpilot/car/toyota/tests/test_legacy_advert_filter.py",
"openpilot/starpilot/car/toyota/tests/test_legacy_advert_callers.py",
"openpilot/starpilot/car/toyota/tests/test_legacy_advert_native.py",
@@ -191,6 +195,7 @@ DEBUG_SAFETY_CLASS_TARGETS = {
RELEASE_TARGETS = (
"opendbc_repo/opendbc/safety/tests/test_toyota_prius_advert.py",
'opendbc_repo/opendbc/safety/tests/test_hyundai_gv60_long.py::TestGV60LongReleaseDenied',
'openpilot/starpilot/tests/test_ford_configuration_parity.py',
'openpilot/starpilot/tests/test_ford_configuration_admission.py',