EV6

opendbc_repo/opendbc/car/hyundai/carcontroller.py

@@ -1033,7 +1033,8 @@ class CarController(CarControllerBase):
             left_sound_active=left_warning.sound_active, right_sound_active=right_warning.sound_active,
           )
         else:
-          adrv_messages = hyundaicanfd.create_adrv_messages(self.packer, self.CAN, self.frame)
+          adrv_messages = hyundaicanfd.create_adrv_messages(self.packer, self.CAN, self.frame,
+                                                             car_fingerprint=self.CP.carFingerprint)
         can_sends.extend(adrv_messages)
         # The front radar treats ADAS_DRV's 0x100 broadcast as its host heartbeat
         # and stops publishing object tracks when it disappears.

opendbc_repo/opendbc/car/hyundai/hyundaicanfd.py

@@ -8,6 +8,33 @@ from opendbc.car.crc import CRC16_XMODEM
 from opendbc.car.hyundai.values import HyundaiFlags, CAR, CANFD_ALT_BUTTONS_RESUME_CAR
 
 
+_adrv_0x51_templates: dict[CAR, bytes] = {}
+
+
+def cache_adrv_0x51_template(car_fingerprint: CAR, dat: bytes | None) -> None:
+  if car_fingerprint != CAR.KIA_EV6:
+    return
+
+  if dat is None:
+    _adrv_0x51_templates.pop(car_fingerprint, None)
+  elif len(dat) == 32 and any(dat[3:]):
+    _adrv_0x51_templates[car_fingerprint] = bytes(dat)
+
+
+def create_adrv_0x51(packer, CAN, frame: int, car_fingerprint: CAR | None = None):
+  template = _adrv_0x51_templates.get(car_fingerprint)
+  if template is None:
+    return packer.make_can_msg("ADRV_0x51", CAN.ACAN, {})
+
+  # EV6 MRR30 tracks stop when the ADAS takeover replaces this platform payload with zeros.
+  dat = bytearray(template)
+  dat[2] = (template[2] + frame + 1) & 0xFF
+  crc = hkg_can_fd_checksum(0x51, None, dat)
+  dat[0] = crc & 0xFF
+  dat[1] = (crc >> 8) & 0xFF
+  return CanData(0x51, bytes(dat), CAN.ACAN)
+
+
 def _set_value(msg: bytearray, sig, ival: int) -> None:
   i = sig.lsb // 8
   bits = sig.size
@@ -788,15 +815,13 @@ def create_fca_warning_light(packer, CAN, frame):
   return ret
 
 
-def create_adrv_messages(packer, CAN, frame, blended_hda2=False):
+def create_adrv_messages(packer, CAN, frame, blended_hda2=False, car_fingerprint=None):
   # messages needed to car happy after disabling
   # the ADAS Driving ECU to do longitudinal control
 
   ret = []
 
-  values = {
-  }
-  ret.append(packer.make_can_msg("ADRV_0x51", CAN.ACAN, values))
+  ret.append(create_adrv_0x51(packer, CAN, frame, car_fingerprint))
 
   if blended_hda2:
     return ret

opendbc_repo/opendbc/car/hyundai/interface.py

@@ -2,6 +2,7 @@ import time
 # Provenance: portions of HKG angle integration are adapted from sunnypilot/opendbc's
 # hkg-angle-steering-2025 branch at cc4b08625. See CREDITS.md and THIRD_PARTY_NOTICES.md.
 from opendbc.car import get_safety_config, structs, uds
+from opendbc.car.hyundai import hyundaicanfd
 from opendbc.car.hyundai.hyundaicanfd import CanBus
 from opendbc.car.hyundai.values import HyundaiFlags, CAR, CarControllerParams, \
                                                    CANFD_UNSUPPORTED_LONGITUDINAL_CAR, \
@@ -391,11 +392,25 @@ class CarInterface(CarInterfaceBase):
         skip_disable_ecu = True
 
       if not skip_disable_ecu:
+        disable_can_recv = can_recv
+        if CP.carFingerprint == CAR.KIA_EV6 and can_recv is not None:
+          hyundaicanfd.cache_adrv_0x51_template(CP.carFingerprint, None)
+          base_can_recv = can_recv
+          adrv_bus = CanBus(CP).ACAN
+
+          def disable_can_recv(*args, **kwargs):
+            packets = base_can_recv(*args, **kwargs)
+            for packet in packets or []:
+              for msg in packet:
+                if msg.src == adrv_bus and msg.address == 0x51:
+                  hyundaicanfd.cache_adrv_0x51_template(CP.carFingerprint, msg.dat)
+            return packets
+
         # Try ECU disable. If it succeeds (IGN-ON mode), enable longitudinal.
         # If it fails (READY mode returns NRC 0x22, or timeout), strip LONG safety flag
         # so panda forwards stock SCC messages normally (lateral-only mode).
         ecu_log(f"=== ECU DISABLE attempt: addr=0x{addr:x}, bus={bus} ===")
-        ecu_disabled = disable_ecu(can_recv, can_send, bus=bus, addr=addr, com_cont_req=communication_control,
+        ecu_disabled = disable_ecu(disable_can_recv, can_send, bus=bus, addr=addr, com_cont_req=communication_control,
                                    reset=bool(CP.flags & HyundaiFlags.CAN_CANFD_BLENDED))
 
         if CP.carFingerprint in (CAR.HYUNDAI_IONIQ_6, CAR.HYUNDAI_IONIQ_5_PE):

opendbc_repo/opendbc/car/hyundai/tests/test_hyundai.py

@@ -149,6 +149,56 @@ class TestHyundaiFingerprint:
 
     assert get_communication_control_request(CAR.HYUNDAI_IONIQ_6) == radar_keepalive_request
 
+  def test_ev6_adrv_0x51_replays_factory_payload(self):
+    CP = CarParams.new_message()
+    CP.carFingerprint = CAR.KIA_EV6
+    CP.flags = int(HyundaiFlags.CANFD | HyundaiFlags.CANFD_LKA_STEERING | HyundaiFlags.EV)
+    packer = CANPacker(DBC[CP.carFingerprint][Bus.pt])
+    can_bus = CanBus(CP)
+    factory = bytes.fromhex("88ed2e091700ffff5e0d0000012006ff021c2200000000000800000010000000")
+
+    hyundaicanfd.cache_adrv_0x51_template(CAR.KIA_EV6, factory)
+    try:
+      address, dat, bus = hyundaicanfd.create_adrv_0x51(packer, can_bus, 7, CAR.KIA_EV6)
+      _, other_dat, _ = hyundaicanfd.create_adrv_0x51(packer, can_bus, 7, CAR.HYUNDAI_IONIQ_6)
+    finally:
+      hyundaicanfd.cache_adrv_0x51_template(CAR.KIA_EV6, None)
+
+    assert address == 0x51
+    assert bus == can_bus.ACAN
+    assert dat[2] == (factory[2] + 8) & 0xFF
+    assert dat[3:] == factory[3:]
+    assert int.from_bytes(dat[:2], "little") == hkg_can_fd_checksum(address, None, bytearray(dat))
+    assert other_dat[3:] == bytes(29)
+
+  def test_ev6_init_captures_factory_adrv_0x51(self, monkeypatch):
+    fingerprint = gen_empty_fingerprint()
+    fingerprint[CanBus(None, fingerprint).CAM][0x50] = 16
+    radar_config = get_radar_track_config(CAR.KIA_EV6)
+    fingerprint[radar_config.bus][radar_config.start_addr] = radar_config.expected_length
+    car_fw = [CarParams.CarFw(ecu=Ecu.adas, fwVersion=b"", address=0x730, brand="hyundai")]
+    CP = CarInterface.get_params(CAR.KIA_EV6, fingerprint, car_fw, True, False, False, get_test_toggles())
+    factory = bytes.fromhex("6b657d090900e1ff000000000020ffff00000000000000000800000010000000")
+
+    def can_recv(*, wait_for_one=True):
+      msg = SimpleNamespace(address=0x51, src=CanBus(CP).ACAN, dat=factory)
+      return [[msg]]
+
+    def fake_disable_ecu(capturing_can_recv, *_args, **_kwargs):
+      capturing_can_recv(wait_for_one=True)
+      return True
+
+    monkeypatch.setattr("opendbc.car.hyundai.interface.disable_ecu", fake_disable_ecu)
+    CarInterface.init(CP, can_recv, None)
+
+    packer = CANPacker(DBC[CP.carFingerprint][Bus.pt])
+    try:
+      _, dat, _ = hyundaicanfd.create_adrv_0x51(packer, CanBus(CP), 0, CAR.KIA_EV6)
+    finally:
+      hyundaicanfd.cache_adrv_0x51_template(CAR.KIA_EV6, None)
+
+    assert dat[3:] == factory[3:]
+
   def test_carnival_hev_low_speed_torque_rate_limits(self):
     CP = CarInterface.get_params(CAR.KIA_CARNIVAL_HEV_4TH_GEN, gen_empty_fingerprint(), [],
                                  False, False, False, None)