Honda: add CanBus class (#31528)

* init canbus class * zero * put back after rebase * cmt * ordering * fix bsm bus --------- Co-authored-by: Shane Smiskol <shane@smiskol.com>
2026-06-20 13:32:04 +08:00 · 2024-02-28 20:45:48 -08:00
parent 7e7a6b1526
commit e0a80d34d9
4 changed files with 59 additions and 39 deletions
@@ -110,6 +110,7 @@ class CarController(CarControllerBase):
    self.CP = CP
    self.packer = CANPacker(dbc_name)
    self.params = CarControllerParams(CP)
+    self.CAN = hondacan.CanBus(CP)
    self.frame = 0

    self.braking = False
@@ -168,7 +169,7 @@ class CarController(CarControllerBase):
        can_sends.append((0x18DAB0F1, 0, b"\x02\x3E\x80\x00\x00\x00\x00\x00", 1))

    # Send steering command.
-    can_sends.append(hondacan.create_steering_control(self.packer, apply_steer, CC.latActive, self.CP.carFingerprint,
+    can_sends.append(hondacan.create_steering_control(self.packer, self.CAN, apply_steer, CC.latActive, self.CP.carFingerprint,
                                                      CS.CP.openpilotLongitudinalControl))

    # wind brake from air resistance decel at high speed
@@ -202,12 +203,12 @@ class CarController(CarControllerBase):

    if not self.CP.openpilotLongitudinalControl:
      if self.frame % 2 == 0 and self.CP.carFingerprint not in HONDA_BOSCH_RADARLESS:  # radarless cars don't have supplemental message
-        can_sends.append(hondacan.create_bosch_supplemental_1(self.packer, self.CP.carFingerprint))
+        can_sends.append(hondacan.create_bosch_supplemental_1(self.packer, self.CAN, self.CP.carFingerprint))
      # If using stock ACC, spam cancel command to kill gas when OP disengages.
      if pcm_cancel_cmd:
-        can_sends.append(hondacan.spam_buttons_command(self.packer, CruiseButtons.CANCEL, self.CP.carFingerprint))
+        can_sends.append(hondacan.spam_buttons_command(self.packer, self.CAN, CruiseButtons.CANCEL, self.CP.carFingerprint))
      elif CC.cruiseControl.resume:
-        can_sends.append(hondacan.spam_buttons_command(self.packer, CruiseButtons.RES_ACCEL, self.CP.carFingerprint))
+        can_sends.append(hondacan.spam_buttons_command(self.packer, self.CAN, CruiseButtons.RES_ACCEL, self.CP.carFingerprint))

    else:
      # Send gas and brake commands.
@@ -220,7 +221,7 @@ class CarController(CarControllerBase):

          stopping = actuators.longControlState == LongCtrlState.stopping
          self.stopping_counter = self.stopping_counter + 1 if stopping else 0
-          can_sends.extend(hondacan.create_acc_commands(self.packer, CC.enabled, CC.longActive, self.accel, self.gas,
+          can_sends.extend(hondacan.create_acc_commands(self.packer, self.CAN, CC.enabled, CC.longActive, self.accel, self.gas,
                                                        self.stopping_counter, self.CP.carFingerprint))
        else:
          apply_brake = clip(self.brake_last - wind_brake, 0.0, 1.0)
@@ -228,7 +229,7 @@ class CarController(CarControllerBase):
          pump_on, self.last_pump_ts = brake_pump_hysteresis(apply_brake, self.apply_brake_last, self.last_pump_ts, ts)

          pcm_override = True
-          can_sends.append(hondacan.create_brake_command(self.packer, apply_brake, pump_on,
+          can_sends.append(hondacan.create_brake_command(self.packer, self.CAN, apply_brake, pump_on,
                                                         pcm_override, pcm_cancel_cmd, fcw_display,
                                                         self.CP.carFingerprint, CS.stock_brake))
          self.apply_brake_last = apply_brake
@@ -251,7 +252,7 @@ class CarController(CarControllerBase):
    if self.frame % 10 == 0:
      hud = HUDData(int(pcm_accel), int(round(hud_v_cruise)), hud_control.leadVisible,
                    hud_control.lanesVisible, fcw_display, acc_alert, steer_required)
-      can_sends.extend(hondacan.create_ui_commands(self.packer, self.CP, CC.enabled, pcm_speed, hud, CS.is_metric, CS.acc_hud, CS.lkas_hud))
+      can_sends.extend(hondacan.create_ui_commands(self.packer, self.CAN, self.CP, CC.enabled, pcm_speed, hud, CS.is_metric, CS.acc_hud, CS.lkas_hud))

      if self.CP.openpilotLongitudinalControl and self.CP.carFingerprint not in HONDA_BOSCH:
        self.speed = pcm_speed
@@ -5,7 +5,7 @@ from openpilot.common.conversions import Conversions as CV
 from openpilot.common.numpy_fast import interp
 from opendbc.can.can_define import CANDefine
 from opendbc.can.parser import CANParser
-from openpilot.selfdrive.car.honda.hondacan import get_cruise_speed_conversion, get_pt_bus
+from openpilot.selfdrive.car.honda.hondacan import CanBus, get_cruise_speed_conversion
 from openpilot.selfdrive.car.honda.values import CAR, DBC, STEER_THRESHOLD, HONDA_BOSCH, \
                                                 HONDA_NIDEC_ALT_SCM_MESSAGES, HONDA_BOSCH_RADARLESS, \
                                                 HondaFlags
@@ -267,7 +267,7 @@ class CarState(CarStateBase):

  def get_can_parser(self, CP):
    messages = get_can_messages(CP, self.gearbox_msg)
-    return CANParser(DBC[CP.carFingerprint]["pt"], messages, get_pt_bus(CP.carFingerprint))
+    return CANParser(DBC[CP.carFingerprint]["pt"], messages, CanBus(CP).pt)

  @staticmethod
  def get_cam_can_parser(CP):
@@ -287,7 +287,7 @@ class CarState(CarStateBase):
        ("BRAKE_COMMAND", 50),
      ]

-    return CANParser(DBC[CP.carFingerprint]["pt"], messages, 2)
+    return CANParser(DBC[CP.carFingerprint]["pt"], messages, CanBus(CP).camera)

  @staticmethod
  def get_body_can_parser(CP):
@@ -296,6 +296,6 @@ class CarState(CarStateBase):
        ("BSM_STATUS_LEFT", 3),
        ("BSM_STATUS_RIGHT", 3),
      ]
-      bus_body = 0 # B-CAN is forwarded to ACC-CAN radar side (CAN 0 on fake ethernet port)
+      bus_body = CanBus(CP).radar # B-CAN is forwarded to ACC-CAN radar side (CAN 0 on fake ethernet port)
      return CANParser(DBC[CP.carFingerprint]["body"], messages, bus_body)
    return None
@@ -1,4 +1,5 @@
 from openpilot.common.conversions import Conversions as CV
+from openpilot.selfdrive.car import CanBusBase
 from openpilot.selfdrive.car.honda.values import HondaFlags, HONDA_BOSCH, HONDA_BOSCH_RADARLESS, CAR, CarControllerParams

 # CAN bus layout with relay
@@ -8,15 +9,34 @@ from openpilot.selfdrive.car.honda.values import HondaFlags, HONDA_BOSCH, HONDA_
 # 3 = F-CAN A - OBDII port


-def get_pt_bus(car_fingerprint):
-  return 1 if car_fingerprint in (HONDA_BOSCH - HONDA_BOSCH_RADARLESS) else 0
+class CanBus(CanBusBase):
+  def __init__(self, CP=None, fingerprint=None) -> None:
+    # use fingerprint if specified
+    super().__init__(CP if fingerprint is None else None, fingerprint)
+
+    if CP.carFingerprint in (HONDA_BOSCH - HONDA_BOSCH_RADARLESS):
+      self._pt, self._radar = self.offset + 1, self.offset
+    else:
+      self._pt, self._radar = self.offset, self.offset + 1
+
+  @property
+  def pt(self) -> int:
+    return self._pt
+
+  @property
+  def radar(self) -> int:
+    return self._radar
+
+  @property
+  def camera(self) -> int:
+    return self.offset + 2


-def get_lkas_cmd_bus(car_fingerprint, radar_disabled=False):
+def get_lkas_cmd_bus(CAN, car_fingerprint, radar_disabled=False):
  no_radar = car_fingerprint in HONDA_BOSCH_RADARLESS
  if radar_disabled or no_radar:
    # when radar is disabled, steering commands are sent directly to powertrain bus
-    return get_pt_bus(car_fingerprint)
+    return CAN.pt
  # normally steering commands are sent to radar, which forwards them to powertrain bus
  return 0

@@ -26,7 +46,7 @@ def get_cruise_speed_conversion(car_fingerprint: str, is_metric: bool) -> float:
  return CV.MPH_TO_MS if car_fingerprint in HONDA_BOSCH_RADARLESS and not is_metric else CV.KPH_TO_MS


-def create_brake_command(packer, apply_brake, pump_on, pcm_override, pcm_cancel_cmd, fcw, car_fingerprint, stock_brake):
+def create_brake_command(packer, CAN, apply_brake, pump_on, pcm_override, pcm_cancel_cmd, fcw, car_fingerprint, stock_brake):
  # TODO: do we loose pressure if we keep pump off for long?
  brakelights = apply_brake > 0
  brake_rq = apply_brake > 0
@@ -47,13 +67,11 @@ def create_brake_command(packer, apply_brake, pump_on, pcm_override, pcm_cancel_
    "AEB_REQ_2": 0,
    "AEB_STATUS": 0,
  }
-  bus = get_pt_bus(car_fingerprint)
-  return packer.make_can_msg("BRAKE_COMMAND", bus, values)
+  return packer.make_can_msg("BRAKE_COMMAND", CAN.pt, values)


-def create_acc_commands(packer, enabled, active, accel, gas, stopping_counter, car_fingerprint):
+def create_acc_commands(packer, CAN, enabled, active, accel, gas, stopping_counter, car_fingerprint):
  commands = []
-  bus = get_pt_bus(car_fingerprint)
  min_gas_accel = CarControllerParams.BOSCH_GAS_LOOKUP_BP[0]

  control_on = 5 if enabled else 0
@@ -90,37 +108,36 @@ def create_acc_commands(packer, enabled, active, accel, gas, stopping_counter, c
      "SET_TO_75": 0x75,
      "SET_TO_30": 0x30,
    }
-    commands.append(packer.make_can_msg("ACC_CONTROL_ON", bus, acc_control_on_values))
+    commands.append(packer.make_can_msg("ACC_CONTROL_ON", CAN.pt, acc_control_on_values))

-  commands.append(packer.make_can_msg("ACC_CONTROL", bus, acc_control_values))
+  commands.append(packer.make_can_msg("ACC_CONTROL", CAN.pt, acc_control_values))
  return commands


-def create_steering_control(packer, apply_steer, lkas_active, car_fingerprint, radar_disabled):
+def create_steering_control(packer, CAN, apply_steer, lkas_active, car_fingerprint, radar_disabled):
  values = {
    "STEER_TORQUE": apply_steer if lkas_active else 0,
    "STEER_TORQUE_REQUEST": lkas_active,
  }
-  bus = get_lkas_cmd_bus(car_fingerprint, radar_disabled)
+  bus = get_lkas_cmd_bus(CAN, car_fingerprint, radar_disabled)
  return packer.make_can_msg("STEERING_CONTROL", bus, values)


-def create_bosch_supplemental_1(packer, car_fingerprint):
+def create_bosch_supplemental_1(packer, CAN, car_fingerprint):
  # non-active params
  values = {
    "SET_ME_X04": 0x04,
    "SET_ME_X80": 0x80,
    "SET_ME_X10": 0x10,
  }
-  bus = get_lkas_cmd_bus(car_fingerprint)
+  bus = get_lkas_cmd_bus(CAN, car_fingerprint)
  return packer.make_can_msg("BOSCH_SUPPLEMENTAL_1", bus, values)


-def create_ui_commands(packer, CP, enabled, pcm_speed, hud, is_metric, acc_hud, lkas_hud):
+def create_ui_commands(packer, CAN, CP, enabled, pcm_speed, hud, is_metric, acc_hud, lkas_hud):
  commands = []
-  bus_pt = get_pt_bus(CP.carFingerprint)
  radar_disabled = CP.carFingerprint in (HONDA_BOSCH - HONDA_BOSCH_RADARLESS) and CP.openpilotLongitudinalControl
-  bus_lkas = get_lkas_cmd_bus(CP.carFingerprint, radar_disabled)
+  bus_lkas = get_lkas_cmd_bus(CAN, CP.carFingerprint, radar_disabled)

  if CP.openpilotLongitudinalControl:
    acc_hud_values = {
@@ -144,7 +161,7 @@ def create_ui_commands(packer, CP, enabled, pcm_speed, hud, is_metric, acc_hud,
      acc_hud_values['FCM_OFF_2'] = acc_hud['FCM_OFF_2']
      acc_hud_values['FCM_PROBLEM'] = acc_hud['FCM_PROBLEM']
      acc_hud_values['ICONS'] = acc_hud['ICONS']
-    commands.append(packer.make_can_msg("ACC_HUD", bus_pt, acc_hud_values))
+    commands.append(packer.make_can_msg("ACC_HUD", CAN.pt, acc_hud_values))

  lkas_hud_values = {
    'SET_ME_X41': 0x41,
@@ -173,19 +190,19 @@ def create_ui_commands(packer, CP, enabled, pcm_speed, hud, is_metric, acc_hud,
      'CMBS_OFF': 0x01,
      'SET_TO_1': 0x01,
    }
-    commands.append(packer.make_can_msg('RADAR_HUD', bus_pt, radar_hud_values))
+    commands.append(packer.make_can_msg('RADAR_HUD', CAN.pt, radar_hud_values))

    if CP.carFingerprint == CAR.CIVIC_BOSCH:
-      commands.append(packer.make_can_msg("LEGACY_BRAKE_COMMAND", bus_pt, {}))
+      commands.append(packer.make_can_msg("LEGACY_BRAKE_COMMAND", CAN.pt, {}))

  return commands


-def spam_buttons_command(packer, button_val, car_fingerprint):
+def spam_buttons_command(packer, CAN, button_val, car_fingerprint):
  values = {
    'CRUISE_BUTTONS': button_val,
    'CRUISE_SETTING': 0,
  }
  # send buttons to camera on radarless cars
-  bus = 2 if car_fingerprint in HONDA_BOSCH_RADARLESS else get_pt_bus(car_fingerprint)
+  bus = CAN.camera if car_fingerprint in HONDA_BOSCH_RADARLESS else CAN.pt
  return packer.make_can_msg("SCM_BUTTONS", bus, values)
@@ -3,7 +3,7 @@ from cereal import car
 from panda import Panda
 from openpilot.common.conversions import Conversions as CV
 from openpilot.common.numpy_fast import interp
-from openpilot.selfdrive.car.honda.hondacan import get_pt_bus
+from openpilot.selfdrive.car.honda.hondacan import CanBus
 from openpilot.selfdrive.car.honda.values import CarControllerParams, CruiseButtons, HondaFlags, CAR, HONDA_BOSCH, HONDA_NIDEC_ALT_SCM_MESSAGES, \
                                                 HONDA_BOSCH_RADARLESS
 from openpilot.selfdrive.car import create_button_events, get_safety_config
@@ -36,6 +36,8 @@ class CarInterface(CarInterfaceBase):
  def _get_params(ret, candidate, fingerprint, car_fw, experimental_long, docs):
    ret.carName = "honda"

+    CAN = CanBus(ret, fingerprint)
+
    if candidate in HONDA_BOSCH:
      ret.safetyConfigs = [get_safety_config(car.CarParams.SafetyModel.hondaBosch)]
      ret.radarUnavailable = True
@@ -47,20 +49,20 @@ class CarInterface(CarInterfaceBase):
      ret.pcmCruise = not ret.openpilotLongitudinalControl
    else:
      ret.safetyConfigs = [get_safety_config(car.CarParams.SafetyModel.hondaNidec)]
-      ret.enableGasInterceptor = 0x201 in fingerprint[0]
+      ret.enableGasInterceptor = 0x201 in fingerprint[CAN.pt]
      ret.openpilotLongitudinalControl = True

      ret.pcmCruise = not ret.enableGasInterceptor

    if candidate == CAR.CRV_5G:
-      ret.enableBsm = 0x12f8bfa7 in fingerprint[0]
+      ret.enableBsm = 0x12f8bfa7 in fingerprint[CAN.radar]

    # Detect Bosch cars with new HUD msgs
    if any(0x33DA in f for f in fingerprint.values()):
      ret.flags |= HondaFlags.BOSCH_EXT_HUD.value

    # Accord ICE 1.5T CVT has different gearbox message
-    if candidate == CAR.ACCORD and 0x191 in fingerprint[1]:
+    if candidate == CAR.ACCORD and 0x191 in fingerprint[CAN.pt]:
      ret.transmissionType = TransmissionType.cvt

    # Certain Hondas have an extra steering sensor at the bottom of the steering rack,
@@ -276,7 +278,7 @@ class CarInterface(CarInterfaceBase):
      raise ValueError(f"unsupported car {candidate}")

    # These cars use alternate user brake msg (0x1BE)
-    if 0x1BE in fingerprint[get_pt_bus(candidate)] and candidate in HONDA_BOSCH:
+    if 0x1BE in fingerprint[CAN.pt] and candidate in HONDA_BOSCH:
      ret.flags |= HondaFlags.BOSCH_ALT_BRAKE.value
      ret.safetyConfigs[0].safetyParam |= Panda.FLAG_HONDA_ALT_BRAKE