Support EPS73 Prius four-byte CAN filters

opendbc_repo/opendbc/car/toyota/carstate.py

@@ -34,7 +34,8 @@ class CarState(CarStateBase):
     super().__init__(CP)
     can_define = CANDefine(DBC[CP.carFingerprint][Bus.pt])
     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.eps_torque_scale = ((CP.safetyConfigs[0].safetyParam & 0xFF) / 100. if prius_advert_filter.configured(CP) else
+                             .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.
 

opendbc_repo/opendbc/car/toyota/interface.py

@@ -77,7 +77,10 @@ class CarInterface(CarInterfaceBase):
       ret.steerActuatorDelay = 0.12  # Default delay, Prius has larger delay
       ret.steerLimitTimer = 0.4
 
-    smart_dsu = (candidate in (CAR.TOYOTA_PRIUS, CAR.TOYOTA_PRIUS_RETROFIT) and 0x2FF in fingerprint[0] and
+    from opendbc.car.toyota import prius_retrofit_advert
+    retrofit_advert = prius_retrofit_advert.hardware_supported(candidate, fingerprint, car_fw)
+    smart_dsu = (candidate in (CAR.TOYOTA_PRIUS, CAR.TOYOTA_PRIUS_RETROFIT) and
+                 (fingerprint.get(0, {}).get(0x2FF) == 8 or retrofit_advert) and
                  any(fw.ecu == Ecu.eps and bytes(fw.fwVersion) == b"8965B47070\x00\x00\x00\x00\x00\x00"
                      for fw in car_fw))
     if candidate in (CAR.TOYOTA_PRIUS, CAR.TOYOTA_PRIUS_RETROFIT) and fingerprint.get(0, {}).get(0x23) == 8:
@@ -162,7 +165,8 @@ class CarInterface(CarInterfaceBase):
       if late_camera:
         bypass = ((0x343 in camera_fp and not any(0x343 in fingerprint.get(bus, {}) for bus in (0, 1))) or
                   (0x4CB in camera_fp and 0x4CB not in fingerprint[0]))
-      smart_dsu = (candidate == CAR.TOYOTA_PRIUS_RETROFIT and 0x2FF in fingerprint[0] and
+      smart_dsu = (candidate == CAR.TOYOTA_PRIUS_RETROFIT and
+                   (fingerprint.get(0, {}).get(0x2FF) == 8 or retrofit_advert) and
                    any(fw.ecu == Ecu.eps and bytes(fw.fwVersion) == b"8965B47070\x00\x00\x00\x00\x00\x00"
                        for fw in car_fw))
       takeover = 0x2FF in fingerprint[0] or bypass

opendbc_repo/opendbc/car/toyota/prius_advert_filter.py

@@ -10,6 +10,9 @@ RETROFIT_EPS = b'8965B47070\x00\x00\x00\x00\x00\x00'
 
 
 def hardware_supported(candidate, fingerprint, car_fw):
+  from opendbc.car.toyota import prius_retrofit_advert
+  if prius_retrofit_advert.hardware_supported(candidate, fingerprint, car_fw):
+    return True
   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
@@ -32,12 +35,16 @@ def hardware_supported(candidate, fingerprint, car_fw):
 
 def apply(cp):
   cp.flags |= int(ToyotaFlags.HYBRID | ToyotaFlags.LONG_FILTER | ToyotaFlags.RAISED_ACCEL_LIMIT)
-  cp.safetyConfigs[0].safetyParam = LONG_WORD
+  from opendbc.car.toyota import prius_retrofit_advert
+  cp.safetyConfigs[0].safetyParam = prius_retrofit_advert.LONG_WORD if cp.safetyConfigs[0].safetyParam & 0xFF == 73 else LONG_WORD
   cp.openpilotLongitudinalControl = True
   cp.alphaLongitudinalAvailable = False
 
 
 def configured(cp):
+  from opendbc.car.toyota import prius_retrofit_advert
+  if prius_retrofit_advert.configured(cp):
+    return True
   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):
@@ -53,7 +60,8 @@ def prepare_stock(cp):
   if not configured(cp):
     return False
   cp.openpilotLongitudinalControl = False
-  cp.safetyConfigs[0].safetyParam = STOCK_WORD
+  from opendbc.car.toyota import prius_retrofit_advert
+  cp.safetyConfigs[0].safetyParam = prius_retrofit_advert.STOCK_WORD if cp.safetyConfigs[0].safetyParam in prius_retrofit_advert.WORDS else STOCK_WORD
   cp.autoResumeSng = False
   return True
 

opendbc_repo/opendbc/car/toyota/prius_retrofit_advert.py

@@ -0,0 +1,42 @@
+"""EPS73 Prius four-byte filter identity, independent of status and physical MAIN."""
+from opendbc.car import structs
+from opendbc.car.toyota.values import CAR, ToyotaFlags
+
+LONG_WORD = 0xD049
+STOCK_WORD = 0xD249
+WORDS = frozenset((LONG_WORD, STOCK_WORD))
+EPS_FIRMWARE = b'8965B47070' + bytes(6)
+CARS = (CAR.TOYOTA_PRIUS, CAR.TOYOTA_PRIUS_RETROFIT)
+
+
+def hardware_supported(candidate, fingerprint, car_fw):
+  if (candidate not in CARS or fingerprint.get(0, {}).get(0x2FF) != 4 or
+      any(0x2FF in messages and (bus not in (0, 2) or messages[0x2FF] != 4) for bus, messages in fingerprint.items()) 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
+  eps = [fw for fw in car_fw if fw.ecu == structs.CarParams.Ecu.eps]
+  if not eps or any(fw.logging or fw.bus != 0 or fw.brand != 'toyota' or fw.address != 0x7A1 or fw.subAddress or
+                    bytes(fw.fwVersion) != EPS_FIRMWARE for fw in eps):
+    return False
+  from opendbc.car.toyota.fingerprints import FW_VERSIONS
+  known = FW_VERSIONS[CAR.TOYOTA_PRIUS]
+  # Cap'n Proto enum readers compare equal to Ecu constants but have different
+  # hashes. Use the catalog's static enum value for each dictionary lookup.
+  for ecu in (structs.CarParams.Ecu.fwdCamera, structs.CarParams.Ecu.fwdRadar):
+    if 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), ()) for fw in car_fw if fw.ecu == ecu):
+      return False
+  return True
+
+
+def configured(cp):
+  if (cp is None or cp.brand != 'toyota' or cp.carFingerprint not in CARS or cp.notCar or cp.passive or cp.dashcamOnly or
+      cp.secOcRequired or not cp.pcmCruise or cp.steerControlType != structs.CarParams.SteerControlType.torque or
+      len(cp.safetyConfigs) != 1 or cp.alphaLongitudinalAvailable or cp.autoResumeSng):
+    return False
+  flags = int(ToyotaFlags.HYBRID | ToyotaFlags.LONG_FILTER | ToyotaFlags.RAISED_ACCEL_LIMIT)
+  config = cp.safetyConfigs[0]
+  return (cp.flags & ~int(ToyotaFlags.ZSS) == flags and config.safetyModel == structs.CarParams.SafetyModel.toyota and
+          config.safetyParam == (LONG_WORD if cp.openpilotLongitudinalControl else STOCK_WORD))

opendbc_repo/opendbc/safety/modes/toyota_aol.h

@@ -8,8 +8,12 @@ static bool toyota_prius_sdsu(void) {
   return toyota_prius_sdsu_param(current_safety_param);
 }
 
+static bool toyota_prius_retrofit_advert_param(uint16_t param) {
+  return (param == 0xD049U) || (param == 0xD249U);
+}
+
 static bool toyota_prius_advert_param(uint16_t param) {
-  return (param == 0xC042U) || (param == 0xC242U);
+  return (param == 0xC042U) || (param == 0xC242U) || toyota_prius_retrofit_advert_param(param);
 }
 
 static bool toyota_prius_advert(void) {

opendbc_repo/opendbc/safety/tests/test_toyota_prius_retrofit_advert.py

@@ -0,0 +1,150 @@
+"""EPS73 Prius four-byte identity never provides a status or source lease."""
+
+import pytest
+from opendbc.car import structs
+from opendbc.car.toyota import prius_retrofit_advert 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_eps73_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_eps73_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_eps73_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_eps73_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()

openpilot/selfdrive/pandad/tests/test_aol_protocol.cc

@@ -179,7 +179,7 @@ int main() {
   }
 
   for (uint32_t word = 0U; word <= 65535U; ++word) {
-    const bool expected = (word == 0xC042U) || (word == 0xC242U) || (word == 0x8042U) || (word == 0x8242U) || (word == 0xC049U) || (word == 0xC249U) || (word == 73U) || (word == 585U) || (word == 856U) || (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 == 0xD049U) || (word == 0xD249U) || (word == 0xC042U) || (word == 0xC242U) || (word == 0x8042U) || (word == 0x8242U) || (word == 0xC049U) || (word == 0xC249U) || (word == 73U) || (word == 585U) || (word == 856U) || (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);

openpilot/starpilot/car/toyota/aol.py

@@ -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, 0xC042, 0xC242,
+  0x8042, 0x8242, 0xC042, 0xC242, 0xD049, 0xD249,
   0xC049, 0xC249,
   0x9449, 0x9549, 0x9558, 0x9564, 0x9649, 0x9749, 0x9758, 0x9764, 0xD44D, 0xD549, 0xD64D, 0xD749,
   73, 585, 856, 4169, 4681, 2121, 0x9049, 0x9249,

openpilot/starpilot/car/toyota/aol_policy.h

@@ -3,7 +3,7 @@
 #include "selfdrive/pandad/aol_protocol.h"
 
 inline bool toyota_aol_param(uint16_t param) {
-  return (param == 0xC042U || param == 0xC242U || param == 0x8042U || param == 0x8242U || param == 0xC049U) || (param == 0xC249U) || (param == 73U) || (param == 585U) || (param == 856U) || (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 == 0xD049U || param == 0xD249U || param == 0xC042U || param == 0xC242U || param == 0x8042U || param == 0x8242U || param == 0xC049U) || (param == 0xC249U) || (param == 73U) || (param == 585U) || (param == 856U) || (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};

openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_callers.py

@@ -0,0 +1,39 @@
+"""EPS73 actual factory/Card/SelfdriveD/Controls/C++/native cold callers."""
+import struct
+import pytest
+from opendbc.car.toyota import prius_retrofit_advert as owner, prius_advert_filter
+from openpilot.starpilot.car.toyota.tests import test_rav4h_caller_native as helpers
+from openpilot.starpilot.car.toyota.tests.test_prius_retrofit_advert_factory import firmware
+from openpilot.starpilot.car.toyota.tests.test_toyota_cpp_transport import host_heartbeat_bridge as host_heartbeat_bridge
+from opendbc.car import Bus
+
+source_matched_native=helpers.source_matched_native
+daemon_gc_scope=helpers.daemon_gc_scope
+
+
+@pytest.mark.parametrize('candidate',owner.CARS)
+@pytest.mark.parametrize('stock',(False,True))
+def test_eps73_advert_actual_cold_callers(candidate,stock,host_heartbeat_bridge):
+  def physical(packer,**fields):
+    fields['sdsu']=None
+    frames=helpers.physical(packer,**fields)
+    assert len(next(data for address,data,bus in frames if address==0x262 and bus==0))==8
+    frames.extend(((0x2FF,bytes.fromhex('80007409'),0),(0x2FF,bytes.fromhex('80007409'),2)))
+    return frames
+
+  def observe(tick,host,cs,cc,safety,native):
+    parser=host.CI.can_parsers[Bus.pt]
+    assert isinstance(parser,prius_advert_filter.PriusAdvertCANParser)
+    assert 0x2FF not in parser.message_states and 'SDSU' not in parser.vl
+    assert host.CI.CS.eps_torque_scale==.73
+    assert host.CP.steerActuatorDelay==struct.unpack('<f',struct.pack('<f',.14))[0]
+    assert host.CP.lateralTuning.torque.steeringAngleDeadzoneDeg==struct.unpack('<f',struct.pack('<f',.3))[0]
+    assert owner.configured(host.CP)
+    if tick>=220:
+      assert not cs.stockAeb and cc.latActive and cc.longActive==(not stock)
+      assert safety.aol_get_permission_mask()==(1 if stock else 3)
+
+  helpers.run_case(stock,candidate=candidate,bridge=host_heartbeat_bridge,input_clock=True,
+                   sdsu_words=(owner.LONG_WORD,owner.STOCK_WORD),firmware_override=firmware(),
+                   fingerprint_frames=[(0x2FF,bytes(4),0)],manual_selection=True,manual_filter_length=4,
+                   filter_advert=True,physical_frames=physical,on_sample=observe)

openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_factory.py

@@ -0,0 +1,90 @@
+"""EPS73 retrofit filter ownership preserves both existing identities and tunes."""
+import pytest
+from opendbc.car import gen_empty_fingerprint, structs
+from opendbc.car.toyota import prius_retrofit_advert as owner, prius_advert_filter, prius_longitudinal
+from opendbc.car.toyota.interface import CarInterface
+from opendbc.car.toyota.values import CAR
+from openpilot.starpilot.car.toyota import aol
+
+
+def firmware():
+  return [structs.CarParams.CarFw(ecu=structs.CarParams.Ecu.eps, address=0x7A1,
+                                fwVersion=owner.EPS_FIRMWARE, brand='toyota', bus=0)]
+
+
+def factory(candidate, *, length=4, zss=False, alpha=False, fp=None, fw=None):
+  if fp is None:
+    fp=gen_empty_fingerprint()
+    fp[0][0x2FF]=length
+    if zss:
+      fp[0][0x23]=8
+  return CarInterface.get_params(candidate, fp, firmware() if fw is None else fw, alpha, False, False)
+
+
+@pytest.mark.parametrize('candidate', owner.CARS)
+@pytest.mark.parametrize('stock', (False, True))
+@pytest.mark.parametrize('alpha', (False, True))
+@pytest.mark.parametrize('zss', (False, True))
+def test_eps73_fourbyte_factory_preserves_existing_retrofit_tunes(candidate, stock, alpha, zss):
+  cp=factory(candidate, alpha=alpha, zss=zss)
+  before=factory(candidate, length=8, alpha=alpha, zss=zss)
+  assert not cp.dashcamOnly and owner.configured(cp)
+  assert cp.openpilotLongitudinalControl and cp.safetyConfigs[0].safetyParam==owner.LONG_WORD
+  if stock:
+    assert prius_longitudinal.prepare_stock(cp) and prius_longitudinal.prepare_stock(before)
+  assert cp.safetyConfigs[0].safetyParam==(owner.STOCK_WORD if stock else owner.LONG_WORD)
+  assert cp.openpilotLongitudinalControl==(not stock)
+  assert prius_advert_filter.configured(cp) and aol.qualified(cp)
+  actual,baseline=cp.to_dict(),before.to_dict()
+  actual.pop('safetyConfigs')
+  baseline.pop('safetyConfigs')
+  assert actual==baseline, 'Existing EPS73 config and all tunes must be preserved'
+
+
+@pytest.mark.parametrize('candidate', owner.CARS)
+@pytest.mark.parametrize('bad', ('primary7','mirror8','bus1','pedal0','pedal2','twoaa0','twoaa2','eps5','main7','camera7'))
+def test_eps73_fourbyte_rejects_ambiguous_topology(candidate,bad):
+  fp=gen_empty_fingerprint()
+  fp[0][0x2FF]=4
+  changes={'primary7':(0,0x2FF,7),'mirror8':(2,0x2FF,8),'bus1':(1,0x2FF,4),
+           'pedal0':(0,0x201,6),'pedal2':(2,0x201,6),'twoaa0':(0,0x2AA,8),'twoaa2':(2,0x2AA,8),
+           'eps5':(0,0x262,5),'main7':(0,0x1D3,7),'camera7':(2,0x343,7)}
+  bus,address,size=changes[bad]
+  fp[bus][address]=size
+  assert not owner.hardware_supported(candidate,fp,firmware())
+  cp=factory(candidate,fp=fp)
+  assert not owner.configured(cp) and cp.safetyConfigs[0].safetyParam not in owner.WORDS
+
+
+@pytest.mark.parametrize('candidate', owner.CARS)
+@pytest.mark.parametrize('bad', ('logging','bus','brand','address','subAddress','fwVersion','missing'))
+def test_eps73_fourbyte_requires_actual_eps_identity(candidate,bad):
+  fw=firmware()
+  if bad=='missing':
+    fw=[]
+  else:
+    values={'logging':True,'bus':1,'brand':'gm','address':0x7A2,'subAddress':1,'fwVersion':b'unknown'}
+    setattr(fw[0],bad,values[bad])
+  fp=gen_empty_fingerprint()
+  fp[0][0x2FF]=4
+  assert not owner.hardware_supported(candidate,fp,fw)
+  assert not owner.configured(factory(candidate,fp=fp,fw=fw))
+
+
+@pytest.mark.parametrize('candidate', owner.CARS)
+@pytest.mark.parametrize('field,value', (('notCar',True),('passive',True),('dashcamOnly',True),('secOcRequired',True),
+                                      ('pcmCruise',False),('alphaLongitudinalAvailable',True),('autoResumeSng',True),
+                                      ('brand','gm'),('steerControlType',structs.CarParams.SteerControlType.angle),
+                                      ('flags',0),('flags',1<<30),('carFingerprint',CAR.TOYOTA_RAV4H)))
+def test_eps73_fourbyte_refuses_foreign_configuration(candidate,field,value):
+  cp=factory(candidate)
+  setattr(cp,field,value)
+  assert not owner.configured(cp) and not prius_advert_filter.configured(cp)
+
+
+@pytest.mark.parametrize('candidate', owner.CARS)
+@pytest.mark.parametrize('word', (0xC042,0xC242,0xC049,4169,4681))
+def test_eps73_fourbyte_does_not_reuse_another_native_identity(candidate,word):
+  cp=factory(candidate)
+  cp.safetyConfigs[0].safetyParam=word
+  assert not owner.configured(cp)

openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_fw.py

@@ -0,0 +1,39 @@
+"""Full acquired firmware uses static catalog keys, never enum-reader hashes."""
+import pytest
+from opendbc.car import gen_empty_fingerprint, structs
+from opendbc.car.toyota import prius_retrofit_advert as owner
+from opendbc.car.toyota.fingerprints import FW_VERSIONS
+from opendbc.car.toyota.values import CAR
+from openpilot.starpilot.car.toyota.tests.test_prius_retrofit_advert_factory import factory
+
+
+def complete_firmware():
+  result=[]
+  for (ecu,address,subaddress),versions in FW_VERSIONS[CAR.TOYOTA_PRIUS].items():
+    result.append(structs.CarParams.CarFw(ecu=ecu,address=address,subAddress=subaddress or 0,
+                                        fwVersion=owner.EPS_FIRMWARE if ecu==structs.CarParams.Ecu.eps else versions[0],brand='toyota',bus=0))
+  assert any(entry.ecu==structs.CarParams.Ecu.fwdCamera for entry in result)
+  assert any(entry.ecu==structs.CarParams.Ecu.fwdRadar for entry in result)
+  return result
+
+
+@pytest.mark.parametrize('candidate',owner.CARS)
+def test_complete_acquired_firmware_is_admitted(candidate):
+  fp=gen_empty_fingerprint()
+  fp[0][0x2FF]=4
+  fw=complete_firmware()
+  assert owner.hardware_supported(candidate,fp,fw), 'EPS73_COMPLETE_ACQUIRED_FW_NOT_ADMITTED'
+  assert owner.configured(factory(candidate,fp=fp,fw=fw))
+
+
+@pytest.mark.parametrize('candidate',owner.CARS)
+@pytest.mark.parametrize('ecu',(structs.CarParams.Ecu.fwdCamera,structs.CarParams.Ecu.fwdRadar))
+@pytest.mark.parametrize('field,value',(('logging',True),('bus',1),('brand','gm'),('fwVersion',b'unknown')))
+def test_complete_acquired_firmware_still_rejects_invalid_metadata(candidate,ecu,field,value):
+  fw=complete_firmware()
+  entry=next(entry for entry in fw if entry.ecu==ecu)
+  setattr(entry,field,value)
+  fp=gen_empty_fingerprint()
+  fp[0][0x2FF]=4
+  assert not owner.hardware_supported(candidate,fp,fw)
+  assert not owner.configured(factory(candidate,fp=fp,fw=fw))

openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_fw_callers.py

@@ -0,0 +1,40 @@
+"""EPS73 actual factory/Card/SelfdriveD/Controls/C++/native cold callers."""
+import struct
+import pytest
+from opendbc.car.toyota import prius_retrofit_advert as owner, prius_advert_filter
+from openpilot.starpilot.car.toyota.tests import test_rav4h_caller_native as helpers
+from openpilot.starpilot.car.toyota.tests.test_prius_retrofit_advert_fw import complete_firmware
+from opendbc.car.toyota.values import CAR
+from openpilot.starpilot.car.toyota.tests.test_toyota_cpp_transport import host_heartbeat_bridge as host_heartbeat_bridge
+from opendbc.car import Bus
+
+source_matched_native=helpers.source_matched_native
+daemon_gc_scope=helpers.daemon_gc_scope
+
+
+@pytest.mark.parametrize('candidate',owner.CARS)
+@pytest.mark.parametrize('stock',(False,True))
+def test_eps73_advert_complete_firmware_actual_callers(candidate,stock,host_heartbeat_bridge):
+  def physical(packer,**fields):
+    fields['sdsu']=None
+    frames=helpers.physical(packer,**fields)
+    assert len(next(data for address,data,bus in frames if address==0x262 and bus==0))==8
+    frames.extend(((0x2FF,bytes.fromhex('80007409'),0),(0x2FF,bytes.fromhex('80007409'),2)))
+    return frames
+
+  def observe(tick,host,cs,cc,safety,native):
+    parser=host.CI.can_parsers[Bus.pt]
+    assert isinstance(parser,prius_advert_filter.PriusAdvertCANParser)
+    assert 0x2FF not in parser.message_states and 'SDSU' not in parser.vl
+    assert host.CI.CS.eps_torque_scale==.73
+    assert host.CP.steerActuatorDelay==struct.unpack('<f',struct.pack('<f',.14))[0]
+    assert host.CP.lateralTuning.torque.steeringAngleDeadzoneDeg==struct.unpack('<f',struct.pack('<f',.3))[0]
+    assert owner.configured(host.CP)
+    if tick>=220:
+      assert not cs.stockAeb and cc.latActive and cc.longActive==(not stock)
+      assert safety.aol_get_permission_mask()==(1 if stock else 3)
+
+  helpers.run_case(stock,candidate=candidate,bridge=host_heartbeat_bridge,input_clock=True,
+                   sdsu_words=(owner.LONG_WORD,owner.STOCK_WORD),firmware_override=complete_firmware(),
+                   fingerprint_frames=[(0x2FF,bytes(4),0)],manual_selection=candidate==CAR.TOYOTA_PRIUS_RETROFIT,manual_filter_length=4,
+                   filter_advert=True,physical_frames=physical,on_sample=observe)

openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_refusals.py

@@ -0,0 +1,56 @@
+"""Focused current EPS73 joins: physical source refusal/recovery and AOL OFF."""
+import pytest
+from opendbc.car import Bus
+from opendbc.car.toyota import prius_advert_filter, prius_retrofit_advert as owner
+from opendbc.car.toyota.values import CAR
+from openpilot.starpilot.car.toyota.tests import test_rav4h_caller_native as helpers
+from openpilot.starpilot.car.toyota.tests.test_prius_retrofit_advert_fw import complete_firmware
+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
+
+
+@pytest.mark.parametrize('stock',(False,True))
+@pytest.mark.parametrize('fault',('main-drop','eps-checksum','ordinary-aol-off'))
+def test_eps73_current_source_refusal_recovery_and_ordinary(stock,fault,host_heartbeat_bridge):
+  tick=0
+
+  def physical(packer,**fields):
+    nonlocal tick
+    tick+=1
+    fields['sdsu']=None
+    frames=helpers.physical(packer,**fields)
+    if 260<=tick<311:
+      if fault=='main-drop':
+        frames=[f for f in frames if f[0]!=0x1D3 or f[2]!=0]
+      elif fault=='eps-checksum':
+        frames=[(a,data[:-1]+bytes([data[-1]^1]),bus) if a==0x262 and bus==0 else (a,data,bus) for a,data,bus in frames]
+    frames.extend(((0x2FF,bytes.fromhex('80007409'),0),(0x2FF,bytes.fromhex('80007409'),2)))
+    return frames
+
+  def expected_rx(address,data,bus,default,cp):
+    if fault=='eps-checksum' and 260<=tick<311 and address==0x262 and bus==0:
+      assert owner.configured(cp) and len(data)==8
+      return False
+    return default
+
+  def observe(tick,host,cs,cc,safety,native):
+    parser=host.CI.can_parsers[Bus.pt]
+    assert isinstance(parser,prius_advert_filter.PriusAdvertCANParser)
+    assert 'SDSU' not in parser.vl and 0x2FF not in parser.message_states
+    assert host.CI.CS.eps_torque_scale==.73 and owner.configured(host.CP)
+    if fault!='ordinary-aol-off' and 301<=tick<308:
+      assert not cc.latActive and not cc.longActive and safety.aol_get_permission_mask()==0
+    if tick>=480 or fault=='ordinary-aol-off' and tick>=220:
+      assert cs.canValid and not cs.canTimeout and cc.latActive and cc.longActive==(not stock)
+      assert safety.aol_get_permission_mask()==(1 if stock else 3)
+    if fault=='ordinary-aol-off' and tick>=220:
+      assert not host.aol_card_intent.settings.enabled and not host.aol_card_intent.allowed_latch
+      assert cs.cruiseState.enabled and cc.enabled
+
+  helpers.run_case(stock,'nominal' if fault=='ordinary-aol-off' else 'main_lease',candidate=CAR.TOYOTA_PRIUS_RETROFIT,
+                   bridge=host_heartbeat_bridge,input_clock=True,aol=fault!='ordinary-aol-off',
+                   sdsu_words=(owner.LONG_WORD,owner.STOCK_WORD),firmware_override=complete_firmware(),
+                   fingerprint_frames=[(0x2FF,bytes(4),0)],manual_selection=True,manual_filter_length=4,filter_advert=True,
+                   physical_frames=physical,native_rx_expected=expected_rx,on_sample=observe)

openpilot/starpilot/car/toyota/tests/test_rav4h_caller_native.py

@@ -250,8 +250,14 @@ def run_case(stock, scenario='nominal', *, aol=True, candidate=CAR.TOYOTA_RAV4H,
     fw = firmware(candidate) if firmware_override is None else firmware_override
     if manual_selection:
       if manual_filter_length is not None:
-        assert candidate == CAR.TOYOTA_MATRIX_RETROFIT and manual_filter_length in (4, 8)
-        assert not fw and firmware_override == []
+        if candidate == CAR.TOYOTA_MATRIX_RETROFIT:
+          assert manual_filter_length in (4, 8) and not fw and firmware_override == []
+        else:
+          assert candidate in (CAR.TOYOTA_PRIUS, CAR.TOYOTA_PRIUS_RETROFIT) and manual_filter_length == 4
+          observed_eps = [entry for entry in fw if entry.ecu == structs.CarParams.Ecu.eps]
+          assert len(observed_eps) == 1 and observed_eps[0].address == 0x7A1 and observed_eps[0].bus == 0
+          assert bytes(observed_eps[0].fwVersion) == b'8965B47070' + bytes(6)
+          assert all(entry.brand == 'toyota' and not entry.logging and entry.bus == 0 for entry in fw)
         assert fingerprint_frames == [(0x2FF, bytes(manual_filter_length), 0)]
       else:
         assert stock and fingerprint_frames == []

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "270b5ff419c82d41512039c2b6188e14cb3a3a8115b334effb74d2090ef54cb5", "schema_version": 1}
+{"protected_inventory_sha256": "729557bebc08863455ded240ae86c63f8b55097a26418fb841d830d77c54a50a", "schema_version": 1}

tools/ci/run_vehicle_tests.py

@@ -29,6 +29,12 @@ AOL_LIFECYCLE_TARGETS = (
   'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_stock_callers.py',
   'opendbc_repo/opendbc/car/toyota/tests/test_matrix_filter.py',
   'openpilot/starpilot/car/toyota/tests/test_matrix_filter_callers.py',
+  'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_factory.py',
+  'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_callers.py',
+  'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_fw.py',
+  'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_fw_callers.py',
+  'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_refusals.py',
+  'opendbc_repo/opendbc/safety/tests/test_toyota_prius_retrofit_advert.py',
   'openpilot/starpilot/aol/tests/test_crossmake_native_admission.py',
   'openpilot/starpilot/aol/tests/test_hkg_source_policy_cache.py',
   'openpilot/starpilot/aol/tests/test_hkg_deadline_admission_cache.py',
@@ -217,6 +223,12 @@ RELEASE_TARGETS = (
   'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_stock_callers.py',
   'opendbc_repo/opendbc/car/toyota/tests/test_matrix_filter.py',
   'openpilot/starpilot/car/toyota/tests/test_matrix_filter_callers.py',
+  'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_factory.py',
+  'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_callers.py',
+  'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_fw.py',
+  'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_fw_callers.py',
+  'openpilot/starpilot/car/toyota/tests/test_prius_retrofit_advert_refusals.py',
+  'opendbc_repo/opendbc/safety/tests/test_toyota_prius_retrofit_advert.py',
   'opendbc_repo/opendbc/car/hyundai/tests/test_long_cp_copy.py',
   'opendbc_repo/opendbc/car/ford/tests/test_mache_display_handoff.py',
   'opendbc_repo/opendbc/safety/tests/test_ford_mache_display.py',