redneck and g90 attempt lkas

2026-07-13 13:22:22 +08:00 · 2026-05-18 20:21:22 -05:00
parent dbecce602c
commit 123adc7edc
7 changed files with 115 additions and 19 deletions
@@ -8,7 +8,7 @@ from opendbc.car import Bus, create_button_events, structs
 from opendbc.car.common.conversions import Conversions as CV
 from opendbc.car.hyundai.hyundaicanfd import CanBus
 from opendbc.car.hyundai.values import HyundaiFlags, HyundaiStarPilotFlags, CAR, DBC, Buttons, CarControllerParams, \
-                                       hyundai_cancel_button_enables_cruise
+                                       hyundai_cancel_button_enables_cruise, ALT_BUS_LDA_BUTTON_CARS
 from opendbc.car.interfaces import CarStateBase

 ButtonType = structs.CarState.ButtonEvent.Type
@@ -25,6 +25,7 @@ BUTTONS_DICT = {Buttons.RES_ACCEL: ButtonType.accelCruise, Buttons.SET_DECEL: Bu
 IONIQ_6_BLINDSPOT_RIGHT_MASK = 0x08
 IONIQ_6_BLINDSPOT_LEFT_MASK = 0x10
 CANFD_CAMERA_LEAD_MIN_DISTANCE = 0.1
+ALT_BUS_LDA_BUTTON_BURST_DEBOUNCE_NS = int(1.3e9)


 def get_non_scc_cruise_signals(CP) -> tuple[str, str, str, str, str]:
@@ -74,6 +75,8 @@ class CarState(CarStateBase):
    self.cruise_buttons: deque = deque([Buttons.NONE] * PREV_BUTTON_SAMPLES, maxlen=PREV_BUTTON_SAMPLES)
    self.main_buttons: deque = deque([Buttons.NONE] * PREV_BUTTON_SAMPLES, maxlen=PREV_BUTTON_SAMPLES)
    self.lda_button = 0
+    self.lda_button_raw = 0
+    self.lda_button_last_raw_rise_ts_nanos = 0
    self.left_paddle = 0
    self.mode_button = 0
    self.custom_button = 0
@@ -168,9 +171,30 @@ class CarState(CarStateBase):

    return False

+  def create_alt_bus_lda_button_events(self, cp_alt: CANParser) -> list[structs.CarState.ButtonEvent]:
+    raw_lda_button = int(cp_alt.vl["CLU13"]["CF_Clu_LdwsLkasSW"])
+    raw_lda_button_ts_nanos = cp_alt.ts_nanos["CLU13"]["CF_Clu_LdwsLkasSW"]
+    button_events: list[structs.CarState.ButtonEvent] = []
+
+    # The G90 cluster pulses this bit multiple times per physical LKAS press burst.
+    # Collapse each burst into a single synthetic press/release pair so downstream
+    # button actions behave like a normal momentary wheel button.
+    if raw_lda_button and not self.lda_button_raw:
+      if self.lda_button_last_raw_rise_ts_nanos == 0 or \
+          raw_lda_button_ts_nanos - self.lda_button_last_raw_rise_ts_nanos > ALT_BUS_LDA_BUTTON_BURST_DEBOUNCE_NS:
+        button_events = [
+          structs.CarState.ButtonEvent(pressed=True, type=ButtonType.lkas),
+          structs.CarState.ButtonEvent(pressed=False, type=ButtonType.lkas),
+        ]
+      self.lda_button_last_raw_rise_ts_nanos = raw_lda_button_ts_nanos
+
+    self.lda_button_raw = raw_lda_button
+    return button_events
+
  def update(self, can_parsers, starpilot_toggles) -> structs.CarState:
    cp = can_parsers[Bus.pt]
    cp_cam = can_parsers[Bus.cam]
+    cp_alt = can_parsers.get(Bus.alt)

    if self.CP.flags & HyundaiFlags.CANFD:
      return self.update_canfd(can_parsers)
@@ -304,14 +328,18 @@ class CarState(CarStateBase):
    prev_cruise_buttons = self.cruise_buttons[-1]
    prev_main_buttons = self.main_buttons[-1]
    prev_lda_button = self.lda_button
+    lkas_button_events = []
    self.cruise_buttons.extend(cp.vl_all["CLU11"]["CF_Clu_CruiseSwState"])
    self.main_buttons.extend(cp.vl_all["CLU11"]["CF_Clu_CruiseSwMain"])
-    if self.CP.flags & HyundaiFlags.HAS_LDA_BUTTON:
+    if self.CP.carFingerprint in ALT_BUS_LDA_BUTTON_CARS and cp_alt is not None:
+      lkas_button_events = self.create_alt_bus_lda_button_events(cp_alt)
+    elif self.CP.flags & HyundaiFlags.HAS_LDA_BUTTON:
      self.lda_button = cp.vl["BCM_PO_11"]["LDA_BTN"]
+      lkas_button_events = create_button_events(self.lda_button, prev_lda_button, {1: ButtonType.lkas})

    ret.buttonEvents = [*self.create_cruise_button_events(self.cruise_buttons[-1], prev_cruise_buttons),
                        *create_button_events(self.main_buttons[-1], prev_main_buttons, {1: ButtonType.mainCruise}),
-                        *create_button_events(self.lda_button, prev_lda_button, {1: ButtonType.lkas})]
+                        *lkas_button_events]

    ret.blockPcmEnable = not self.recent_button_interaction()

@@ -510,7 +538,10 @@ class CarState(CarStateBase):
    if CP.flags & HyundaiFlags.NON_SCC and not (CP.flags & HyundaiFlags.NON_SCC_NO_FCA):
      msgs.append(("FCA11", 0))  # Non-SCC trims can stop publishing FCA11; don't let it poison canValid

-    return {
+    parsers = {
      Bus.pt: CANParser(DBC[CP.carFingerprint][Bus.pt], msgs, 0),
      Bus.cam: CANParser(DBC[CP.carFingerprint][Bus.pt], [], 2),
    }
+    if CP.carFingerprint in ALT_BUS_LDA_BUTTON_CARS:
+      parsers[Bus.alt] = CANParser(DBC[CP.carFingerprint][Bus.pt], [("CLU13", 0)], 1)
+    return parsers
@@ -5,7 +5,7 @@ from opendbc.car.hyundai.values import HyundaiFlags, CAR, DBC, CarControllerPara
                                                   CANFD_UNSUPPORTED_LONGITUDINAL_CAR, \
                                                   CANFD_SECURITYACCESS_CAR, \
                                                   UNSUPPORTED_LONGITUDINAL_CAR, HyundaiSafetyFlags, \
-                                                   hyundai_cancel_button_enables_cruise
+                                                   hyundai_cancel_button_enables_cruise, ALT_BUS_LDA_BUTTON_CARS
 from opendbc.car.hyundai.radar_interface import RADAR_START_ADDR
 from opendbc.car.interfaces import CarInterfaceBase, ACCEL_MIN
 from opendbc.car.disable_ecu import disable_ecu, ecu_log
@@ -139,7 +139,7 @@ class CarInterface(CarInterfaceBase):
        ret.safetyConfigs[0].safetyParam |= HyundaiSafetyFlags.CAMERA_SCC.value

      # These cars have the LFA button on the steering wheel
-      if 0x391 in fingerprint[0] or ret.flags & HyundaiFlags.CAN_CANFD_BLENDED:
+      if candidate in ALT_BUS_LDA_BUTTON_CARS or 0x391 in fingerprint[0] or ret.flags & HyundaiFlags.CAN_CANFD_BLENDED:
        ret.flags |= HyundaiFlags.HAS_LDA_BUTTON.value
      if ret.flags & HyundaiFlags.CAN_CANFD_BLENDED:
        ret.safetyConfigs[-1].safetyParam |= HyundaiSafetyFlags.CAN_CANFD_BLENDED.value
@@ -10,7 +10,8 @@ from opendbc.car.fw_versions import build_fw_dict, match_fw_to_car
 from opendbc.car.hyundai.carcontroller import CarController, Ioniq6LongitudinalTuningState, GenesisG90LongitudinalTuningState, \
                                             update_ioniq_6_longitudinal_tuning, \
                                             update_genesis_g90_longitudinal_tuning
-from opendbc.car.hyundai.carstate import CarState, decode_canfd_camera_lead, decode_ioniq_6_blindspot_radar_state
+from opendbc.car.hyundai.carstate import CarState, decode_canfd_camera_lead, decode_ioniq_6_blindspot_radar_state, \
+                                         ALT_BUS_LDA_BUTTON_BURST_DEBOUNCE_NS
 from opendbc.car.hyundai.interface import CarInterface
 from opendbc.car.hyundai import hyundaican, hyundaicanfd
 from opendbc.car.hyundai.hyundaicanfd import CanBus
@@ -18,7 +19,8 @@ from opendbc.car.hyundai.radar_interface import RADAR_START_ADDR
 from opendbc.car.hyundai.values import CAMERA_SCC_CAR, CANFD_CAR, CAN_GEARS, CAR, CHECKSUM, DATE_FW_ECUS, \
                                         HYBRID_CAR, EV_CAR, FW_QUERY_CONFIG, LEGACY_SAFETY_MODE_CAR, CANFD_FUZZY_WHITELIST, \
                                         UNSUPPORTED_LONGITUDINAL_CAR, PLATFORM_CODE_ECUS, HYUNDAI_VERSION_REQUEST_LONG, \
-                                         CarControllerParams, DBC, HyundaiFlags, get_platform_codes, HyundaiSafetyFlags, Buttons
+                                         CarControllerParams, DBC, HyundaiFlags, get_platform_codes, HyundaiSafetyFlags, Buttons, \
+                                         HyundaiStarPilotSafetyFlags

 LongCtrlState = CarControl.Actuators.LongControlState
 from opendbc.car.hyundai.fingerprints import FW_VERSIONS
@@ -209,6 +211,47 @@ class TestHyundaiFingerprint:
    assert CP.vEgoStopping == pytest.approx(0.8)
    assert CP.stoppingDecelRate == pytest.approx(0.55)

+  def test_genesis_g90_sets_has_lkas_button_flag(self):
+    toggles = get_test_toggles()
+    CP = CarInterface.get_params(CAR.GENESIS_G90, gen_empty_fingerprint(), [], True, False, False, toggles)
+    FPCP = CarInterface.get_starpilot_params(CAR.GENESIS_G90, gen_empty_fingerprint(), [], CP, toggles)
+
+    assert CP.flags & HyundaiFlags.HAS_LDA_BUTTON
+    assert FPCP.safetyConfigs[-1].safetyParam & HyundaiStarPilotSafetyFlags.HAS_LDA_BUTTON
+
+  def test_genesis_g90_lkas_button_uses_clu13_on_alt_bus(self):
+    toggles = get_test_toggles()
+    CP = CarInterface.get_params(CAR.GENESIS_G90, gen_empty_fingerprint(), [], True, False, False, toggles)
+    FPCP = CarInterface.get_starpilot_params(CAR.GENESIS_G90, gen_empty_fingerprint(), [], CP, toggles)
+
+    car_state = CarState(CP, FPCP)
+    can_parsers = car_state.get_can_parsers(CP)
+    packer = CANPacker(DBC[CP.carFingerprint][Bus.pt])
+
+    assert Bus.alt in can_parsers
+
+    can_parsers[Bus.alt].update([(1, [packer.make_can_msg("CLU13", 1, {"CF_Clu_LdwsLkasSW": 1})])])
+    ret, _ = car_state.update(can_parsers, toggles)
+    assert [(be.type, be.pressed) for be in ret.buttonEvents] == [(ButtonType.lkas, True), (ButtonType.lkas, False)]
+
+    can_parsers[Bus.alt].update([(2, [packer.make_can_msg("CLU13", 1, {"CF_Clu_LdwsLkasSW": 0})])])
+    ret, _ = car_state.update(can_parsers, toggles)
+    assert len(ret.buttonEvents) == 0
+
+    can_parsers[Bus.alt].update([(3, [packer.make_can_msg("CLU13", 1, {"CF_Clu_LdwsLkasSW": 1})])])
+    ret, _ = car_state.update(can_parsers, toggles)
+    assert len(ret.buttonEvents) == 0
+
+    quiet_ts = ALT_BUS_LDA_BUTTON_BURST_DEBOUNCE_NS + 3
+    can_parsers[Bus.alt].update([(quiet_ts, [packer.make_can_msg("CLU13", 1, {"CF_Clu_LdwsLkasSW": 0})])])
+    ret, _ = car_state.update(can_parsers, toggles)
+    assert len(ret.buttonEvents) == 0
+
+    next_press_ts = ALT_BUS_LDA_BUTTON_BURST_DEBOUNCE_NS + 4
+    can_parsers[Bus.alt].update([(next_press_ts, [packer.make_can_msg("CLU13", 1, {"CF_Clu_LdwsLkasSW": 1})])])
+    ret, _ = car_state.update(can_parsers, toggles)
+    assert [(be.type, be.pressed) for be in ret.buttonEvents] == [(ButtonType.lkas, True), (ButtonType.lkas, False)]
+
  def test_palisade_2023_longitudinal_params_soften_final_stop_hold(self):
    toggles = get_test_toggles()
    CP = CarInterface.get_params(CAR.HYUNDAI_PALISADE_2023, gen_empty_fingerprint(), [], True, False, False, toggles)
@@ -759,6 +759,11 @@ CANCEL_BUTTON_ENABLE_CARS = frozenset({
 })


+ALT_BUS_LDA_BUTTON_CARS = frozenset({
+  CAR.GENESIS_G90,
+})
+
+
 def hyundai_cancel_button_enables_cruise(car_fingerprint) -> bool:
  return car_fingerprint in CANCEL_BUTTON_ENABLE_CARS

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+import math
 import os
 import time
 import threading
@@ -74,7 +75,7 @@ class Car:

  def __init__(self, CI=None, RI=None) -> None:
    self.can_sock = messaging.sub_sock('can', timeout=20)
-    self.sm = messaging.SubMaster(['pandaStates', 'carControl', 'onroadEvents', 'radarState'])
+    self.sm = messaging.SubMaster(['pandaStates', 'carControl', 'onroadEvents', 'radarState', 'longitudinalPlan'])
    self.pm = messaging.PubMaster(['sendcan', 'carState', 'carParams', 'carOutput', 'liveTracks'])

    self.can_rcv_cum_timeout_counter = 0
@@ -372,10 +373,20 @@ class Car:
    if self.redneck_cruise is None:
      return

-    send_button, v_target = self.redneck_cruise.run(CS, CC, self.sm['starpilotPlan'].vCruise, self.is_metric)
+    send_button, v_target = self.redneck_cruise.run(CS, CC, self._get_redneck_target_speed(), self.is_metric)
    self.CI.CS.redneck_send_button = send_button
    self.CI.CS.redneck_v_target = v_target

+  def _get_redneck_target_speed(self) -> float:
+    if self.sm.seen['longitudinalPlan'] and self.sm.valid['longitudinalPlan']:
+      speeds = self.sm['longitudinalPlan'].speeds
+      if len(speeds) > 0:
+        target_speed = float(speeds[0])
+        if math.isfinite(target_speed):
+          return target_speed
+
+    return float(self.sm['starpilotPlan'].vCruise)
+
  def step(self):
    CS, RD, FPCS = self.state_update()

@@ -13,12 +13,12 @@ HYST_GAP = 0.0
 INACTIVE_TIMER = 0.4

 CRUISE_BUTTON_TIMERS = {
-  ButtonType.decelCruise: 0,
-  ButtonType.accelCruise: 0,
-  ButtonType.setCruise: 0,
-  ButtonType.resumeCruise: 0,
-  ButtonType.cancel: 0,
-  ButtonType.mainCruise: 0,
+  int(ButtonType.decelCruise): 0,
+  int(ButtonType.accelCruise): 0,
+  int(ButtonType.setCruise): 0,
+  int(ButtonType.resumeCruise): 0,
+  int(ButtonType.cancel): 0,
+  int(ButtonType.mainCruise): 0,
 }


@@ -26,14 +26,15 @@ def get_minimum_set_speed(is_metric: bool) -> int:
  return 30 if is_metric else 20


-def update_manual_button_timers(CS: car.CarState, button_timers: dict[ButtonType, int]) -> None:
+def update_manual_button_timers(CS: car.CarState, button_timers: dict[int, int]) -> None:
  for button_type in button_timers:
    if button_timers[button_type] > 0:
      button_timers[button_type] += 1

  for event in CS.buttonEvents:
-    if event.type in button_timers:
-      button_timers[event.type] = 1 if event.pressed else 0
+    button_type = event.type.raw if hasattr(event.type, "raw") else int(event.type)
+    if button_type in button_timers:
+      button_timers[button_type] = 1 if event.pressed else 0


 class RedneckCruise:
@@ -68,6 +68,11 @@ class TestRedneckCruise(unittest.TestCase):
    send_button, _ = self._run_until_active(target_mph=25.0, speed_cluster_mph=20.0, button_events=[button_event])
    self.assertEqual(SEND_BUTTON_NONE, send_button)

+  def test_suppresses_output_for_capnp_style_button_events(self):
+    button_event = SimpleNamespace(type=SimpleNamespace(raw=int(ButtonType.accelCruise)), pressed=True)
+    send_button, _ = self._run_until_active(target_mph=25.0, speed_cluster_mph=20.0, button_events=[button_event])
+    self.assertEqual(SEND_BUTTON_NONE, send_button)
+
  def test_suppresses_output_for_override_cancel_and_resume(self):
    for kwargs in ({"override": True}, {"cancel": True}, {"resume": True}):
      with self.subTest(**kwargs):