移植 IQ.Pilot panda 子系统

- 用 IQ.Pilot 的 panda/ 完整替换原版（含 board 固件、python 库、certs、构建脚本）
- 同步 selfdrive/pandad/ 为 IQ.Pilot 版（多 panda 管理，新增 panda_comms.cc）
- 修复 IQ.Pilot board 固件编译问题：
  - health.h 补充 CAN_PACKET_VERSION 宏定义
  - IQ.Pilot 特有变量/函数名改回 sunnypilot 原版
    (current_safety_param_iq→current_safety_param_sp 等)
- 固件编译通过：panda(F4)/panda_h7/panda_jungle_h7/body_h7

panda/python/__init__.py

@@ -1,34 +1,23 @@
 # python library to interface with panda
 import os
-import re
 import sys
 import time
 import usb1
 import struct
 import hashlib
 import binascii
-import ctypes
 from functools import wraps, partial
 from itertools import accumulate
 
-import opendbc
 from opendbc.car.structs import CarParams
 
 from .base import BaseHandle
-from .constants import BASEDIR, FW_PATH, McuType, compute_version_hash
+from .constants import FW_PATH, McuType
 from .dfu import PandaDFU
 from .spi import PandaSpiHandle, PandaSpiException, PandaProtocolMismatch
 from .usb import PandaUsbHandle
 from .utils import logger
 
-# load libusb from pip package
-try:
-  import libusb_package
-  usb1._libusb1.loadLibrary(ctypes.CDLL(str(libusb_package.get_library_path())))
-except ImportError:
-  # TODO: remove this on next AGNOS update
-  pass
-
 __version__ = '0.0.10'
 
 CANPACKET_HEAD_SIZE = 0x6
@@ -43,15 +32,6 @@ def calculate_checksum(data):
     res ^= b
   return res
 
-def _parse_c_struct(path, name):
-  type_to_format = {"uint8_t": "B", "uint16_t": "H", "uint32_t": "I", "float": "f"}
-  with open(path) as f:
-    lines = [l.strip() for l in f.read().split(f"struct __attribute__((packed)) {name} {{", 1)[1].split("};", 1)[0].splitlines() if l.strip()]
-  fields = [re.fullmatch(rf"({'|'.join(type_to_format)})\s+\w+;", l) for l in lines]
-  if not all(fields):
-    raise ValueError(f"unsupported {name} layout in {path}")
-  return struct.Struct("<" + "".join(type_to_format[m[1]] for m in fields))
-
 def pack_can_buffer(arr, chunk=False, fd=False):
   snds = [bytearray(), ]
   for address, dat, bus in arr:
@@ -115,9 +95,13 @@ def ensure_version(desc, lib_field, panda_field, fn):
     return fn(self, *args, **kwargs)
   return wrapper
 ensure_can_packet_version = partial(ensure_version, "CAN", "CAN_PACKET_VERSION", "can_version")
+ensure_can_health_packet_version = partial(ensure_version, "CAN health", "CAN_HEALTH_PACKET_VERSION", "can_health_version")
 ensure_health_packet_version = partial(ensure_version, "health", "HEALTH_PACKET_VERSION", "health_version")
 
 
+def _is_lite_hw() -> bool:
+  return os.environ.get('LITE') == '1' or os.path.exists('/tmp/lite_hw')
+
 
 class Panda:
 
@@ -131,20 +115,35 @@ class Panda:
 
   # from https://github.com/commaai/openpilot/blob/103b4df18cbc38f4129555ab8b15824d1a672bdf/cereal/log.capnp#L648
   HW_TYPE_UNKNOWN = b'\x00'
+  HW_TYPE_WHITE = b'\x01'
+  HW_TYPE_GREY_PANDA = b'\x02'
+  HW_TYPE_BLACK = b'\x03'
+  HW_TYPE_PEDAL = b'\x04'
+  HW_TYPE_UNO = b'\x05'
+  HW_TYPE_DOS = b'\x06'
   HW_TYPE_RED_PANDA = b'\x07'
+  HW_TYPE_RED_PANDA_V2 = b'\x08'
   HW_TYPE_TRES = b'\x09'
   HW_TYPE_CUATRO = b'\x0a'
   HW_TYPE_BODY = b'\xb1'
 
-  CAN_PACKET_VERSION = compute_version_hash(os.path.join(opendbc.INCLUDE_PATH, "opendbc/safety/can.h"))
-  HEALTH_PACKET_VERSION = compute_version_hash(os.path.join(BASEDIR, "board/health.h"))
-  HEALTH_STRUCT = _parse_c_struct(os.path.join(BASEDIR, "board/health.h"), "health_t")
+  CAN_PACKET_VERSION = 4
+  HEALTH_PACKET_VERSION = 17
+  CAN_HEALTH_PACKET_VERSION = 5
+  HEALTH_STRUCT = struct.Struct("<IIIIIIIIBBBBBHBBBHfBBHHHB")
   CAN_HEALTH_STRUCT = struct.Struct("<BIBBBBBBBBIIIIIIIHHBBBIIII")
 
+  F4_DEVICES = [HW_TYPE_WHITE, HW_TYPE_BLACK, HW_TYPE_DOS]
   H7_DEVICES = [HW_TYPE_RED_PANDA, HW_TYPE_TRES, HW_TYPE_CUATRO, HW_TYPE_BODY]
-  SUPPORTED_DEVICES = H7_DEVICES
+  SUPPORTED_DEVICES = H7_DEVICES + F4_DEVICES + [HW_TYPE_GREY_PANDA, HW_TYPE_PEDAL, HW_TYPE_UNO, HW_TYPE_RED_PANDA_V2]
 
-  INTERNAL_DEVICES = (HW_TYPE_TRES, HW_TYPE_CUATRO)
+  INTERNAL_DEVICES = (HW_TYPE_DOS, HW_TYPE_TRES, HW_TYPE_CUATRO)
+
+  MAX_FAN_RPMs = {
+    HW_TYPE_DOS: 6500,
+    HW_TYPE_TRES: 6600,
+    HW_TYPE_CUATRO: 12500,
+  }
 
   HARNESS_STATUS_NC = 0
   HARNESS_STATUS_NORMAL = 1
@@ -163,6 +162,7 @@ class Panda:
     else:
       self._connect_serial = serial
 
+    # connect and set mcu type
     self.connect(claim)
 
   def _cli_select_panda(self):
@@ -207,21 +207,34 @@ class Panda:
     self._handle = None
     while self._handle is None:
       # try USB first, then SPI
-      self._context, self._handle, serial, self.bootstub = self.usb_connect(self._connect_serial, claim=claim, no_error=wait)
+      self._context, self._handle, serial, self.bootstub, bcd = self.usb_connect(self._connect_serial, claim=claim, no_error=wait)
       if self._handle is None:
-        self._context, self._handle, serial, self.bootstub = self.spi_connect(self._connect_serial)
+        self._context, self._handle, serial, self.bootstub, bcd = self.spi_connect(self._connect_serial)
       if not wait:
         break
 
     if self._handle is None:
       raise Exception("failed to connect to panda")
 
+    self._bcd_hw_type = None
+    ret = self._handle.controlRead(Panda.REQUEST_IN, 0xc1, 0, 0, 0x40)
+    missing_hw_type_endpoint = self.bootstub and ret.startswith(b'\xff\x00\xc1\x3e\xde\xad\xd0\x0d')
+    if missing_hw_type_endpoint and bcd is not None:
+      self._bcd_hw_type = bcd
+
+    self._assume_f4_mcu = (self._bcd_hw_type is None) and missing_hw_type_endpoint
+
     self._serial = serial
     self._connect_serial = serial
     self._handle_open = True
-    self.health_version, self.can_version = self.get_packets_versions()
+    self._mcu_type = self.get_mcu_type()
+    self.health_version, self.can_version, self.can_health_version = self.get_packets_versions()
     logger.debug("connected")
 
+    hw_type = self.get_type()
+    if hw_type not in self.SUPPORTED_DEVICES:
+      print("WARNING: Using deprecated HW")
+
     # disable openpilot's heartbeat checks
     if self._disable_checks:
       self.set_heartbeat_disabled()
@@ -248,7 +261,7 @@ class Panda:
       handle = PandaSpiHandle()
       dat = handle.get_protocol_version()
     except PandaSpiException:
-      return None, None, None, False
+      return None, None, None, False, None
 
     spi_serial = binascii.hexlify(dat[:12]).decode()
     pid = dat[13]
@@ -259,18 +272,18 @@ class Panda:
 
     # did we get the right panda?
     if serial is not None and spi_serial != serial:
-      return None, None, None, False
+      return None, None, None, False, None
 
     # ensure our protocol version matches the panda
     if (not ignore_version) and spi_version != handle.PROTOCOL_VERSION:
       raise PandaProtocolMismatch(f"panda protocol mismatch: expected {handle.PROTOCOL_VERSION}, got {spi_version}. reflash panda")
 
     # got a device and all good
-    return None, handle, spi_serial, bootstub
+    return None, handle, spi_serial, bootstub, None
 
   @classmethod
   def usb_connect(cls, serial, claim=True, no_error=False):
-    handle, usb_serial, bootstub = None, None, None
+    handle, usb_serial, bootstub, bcd = None, None, None, None
     context = usb1.USBContext()
     context.open()
     try:
@@ -292,10 +305,14 @@ class Panda:
             handle = device.open()
             if sys.platform not in ("win32", "cygwin", "msys", "darwin"):
               handle.setAutoDetachKernelDriver(True)
-            if claim or sys.platform == "darwin":
+            if claim:
               handle.claimInterface(0)
               # handle.setInterfaceAltSetting(0, 0)  # Issue in USB stack
 
+            this_bcd = device.getbcdDevice()
+            if this_bcd is not None and this_bcd != 0x2300:
+              bcd = bytearray([this_bcd >> 8, ])
+
             break
     except Exception:
       logger.exception("USB connect error")
@@ -306,7 +323,7 @@ class Panda:
     else:
       context.close()
 
-    return context, usb_handle, usb_serial, bootstub
+    return context, usb_handle, usb_serial, bootstub, bcd
 
   def is_connected_spi(self):
     return isinstance(self._handle, PandaSpiHandle)
@@ -342,15 +359,12 @@ class Panda:
 
   @classmethod
   def spi_list(cls):
-    _, _, serial, _ = cls.spi_connect(None, ignore_version=True)
+    _, _, serial, _, _ = cls.spi_connect(None, ignore_version=True)
     if serial is not None:
       return [serial, ]
     return []
 
   def reset(self, enter_bootstub=False, enter_bootloader=False, reconnect=True):
-    if enter_bootstub or enter_bootloader:
-      assert (hw_type := self.get_type()) in self.SUPPORTED_DEVICES, f"Unknown HW: {hw_type}"
-
     # no response is expected since it resets right away
     timeout = 5000 if isinstance(self._handle, PandaSpiHandle) else 15000
     try:
@@ -430,14 +444,17 @@ class Panda:
       pass
 
   def flash(self, fn=None, code=None, reconnect=True):
-    assert (hw_type := self.get_type()) in self.SUPPORTED_DEVICES, f"Unknown HW: {hw_type}"
-
     if self.up_to_date(fn=fn):
       logger.info("flash: already up to date")
       return
 
+    hw_type = self.get_type()
+    if hw_type not in self.SUPPORTED_DEVICES:
+      if not (hw_type == Panda.HW_TYPE_UNKNOWN and _is_lite_hw()):
+        raise RuntimeError(f"HW type {hw_type.hex()} is deprecated and can no longer be flashed.")
+
     if not fn:
-      fn = os.path.join(FW_PATH, McuType.H7.config.app_fn)
+      fn = os.path.join(FW_PATH, self._mcu_type.config.app_fn)
     assert os.path.isfile(fn)
     logger.debug("flash: main version is %s", self.get_version())
     if not self.bootstub:
@@ -452,7 +469,7 @@ class Panda:
     logger.debug("flash: bootstub version is %s", self.get_version())
 
     # do flash
-    Panda.flash_static(self._handle, code, mcu_type=McuType.H7)
+    Panda.flash_static(self._handle, code, mcu_type=self._mcu_type)
 
     # reconnect
     if reconnect:
@@ -503,7 +520,7 @@ class Panda:
   def up_to_date(self, fn=None) -> bool:
     current = self.get_signature()
     if fn is None:
-      fn = os.path.join(FW_PATH, McuType.H7.config.app_fn)
+      fn = os.path.join(FW_PATH, self.get_mcu_type().config.app_fn)
     expected = Panda.get_signature_from_firmware(fn)
     return (current == expected)
 
@@ -542,12 +559,9 @@ class Panda:
       "sbu1_voltage_mV": a[22],
       "sbu2_voltage_mV": a[23],
       "som_reset_triggered": a[24],
-      "sound_output_level": a[25],
-      "controls_allowed_lateral": a[26],
-      "controls_allowed_longitudinal": a[27],
     }
 
-  @ensure_health_packet_version
+  @ensure_can_health_packet_version
   def can_health(self, can_number):
     LEC_ERROR_CODE = {
       0: "No error",
@@ -607,16 +621,43 @@ class Panda:
     return bytes(part_1 + part_2)
 
   def get_type(self):
-    return self._handle.controlRead(Panda.REQUEST_IN, 0xc1, 0, 0, 0x40)
+    ret = self._handle.controlRead(Panda.REQUEST_IN, 0xc1, 0, 0, 0x40)
 
+    if self._bcd_hw_type is not None and (ret is None or len(ret) != 1):
+      ret = self._bcd_hw_type
+
+    return ret
+
+  # Returns tuple with health packet version and CAN packet/USB packet version
   def get_packets_versions(self):
-    dat = self._handle.controlRead(Panda.REQUEST_IN, 0xdd, 0, 0, 8)
-    if dat and len(dat) == 8:
-      return struct.unpack("<II", dat)
-    return (0, 0)
+    dat = self._handle.controlRead(Panda.REQUEST_IN, 0xdd, 0, 0, 3)
+    if dat and len(dat) == 3:
+      a = struct.unpack("BBB", dat)
+      return (a[0], a[1], a[2])
+    else:
+      return (0, 0, 0)
+
+  def get_mcu_type(self) -> McuType:
+    hw_type = self.get_type()
+    if hw_type in Panda.F4_DEVICES:
+      return McuType.F4
+    elif hw_type in Panda.H7_DEVICES:
+      return McuType.H7
+    else:
+      if self._assume_f4_mcu:
+        return McuType.F4
+      # C3 Lite (mr.one clone): panda has STM32F4 MCU (16-sector DFU layout = DOS board)
+      if hw_type == Panda.HW_TYPE_UNKNOWN and _is_lite_hw():
+        return McuType.F4
+    raise ValueError(f"unknown HW type: {hw_type}")
 
   def is_internal(self):
-    return self.get_type() in Panda.INTERNAL_DEVICES
+    hw_type = self.get_type()
+    if hw_type in Panda.INTERNAL_DEVICES:
+      return True
+    if hw_type == Panda.HW_TYPE_UNKNOWN and _is_lite_hw():
+      return True
+    return False
 
   def get_serial(self):
     """
@@ -635,7 +676,7 @@ class Panda:
     return self._serial
 
   def get_dfu_serial(self):
-    return PandaDFU.st_serial_to_dfu_serial(self._serial, McuType.H7)
+    return PandaDFU.st_serial_to_dfu_serial(self._serial, self._mcu_type)
 
   def get_uid(self):
     """
@@ -653,15 +694,12 @@ class Panda:
 
   # ******************* configuration *******************
 
-  def set_alternative_experience(self, alternative_experience, safety_param_sp=0):
-    self._handle.controlWrite(Panda.REQUEST_OUT, 0xdf, int(alternative_experience), int(safety_param_sp), b'')
+  def set_alternative_experience(self, alternative_experience, safety_param_iq=0):
+    self._handle.controlWrite(Panda.REQUEST_OUT, 0xdf, int(alternative_experience), int(safety_param_iq), b'')
 
   def set_power_save(self, power_save_enabled=0):
     self._handle.controlWrite(Panda.REQUEST_OUT, 0xe7, int(power_save_enabled), 0, b'')
 
-  def enter_stop_mode(self):
-    self._handle.controlWrite(Panda.REQUEST_OUT, 0xb5, 0, 0, b'', expect_disconnect=True)
-
   def set_safety_mode(self, mode=CarParams.SafetyModel.silent, param=0):
     self._handle.controlWrite(Panda.REQUEST_OUT, 0xdc, mode, param, b'')
 
@@ -762,8 +800,8 @@ class Panda:
       ret += self._handle.bulkWrite(2, struct.pack("B", port_number) + ln[i:i + 0x20])
     return ret
 
-  def send_heartbeat(self, engaged=True, engaged_mads=True):
-    self._handle.controlWrite(Panda.REQUEST_OUT, 0xf3, engaged, engaged_mads, b'')
+  def send_heartbeat(self, engaged=True, engaged_aol=True):
+    self._handle.controlWrite(Panda.REQUEST_OUT, 0xf3, engaged, engaged_aol, b'')
 
   # disable heartbeat checks for use outside of openpilot
   # sending a heartbeat will reenable the checks

panda/python/constants.py

@@ -1,16 +1,10 @@
 import os
 import enum
-import hashlib
 from typing import NamedTuple
 
 BASEDIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "../")
 FW_PATH = os.path.join(BASEDIR, "board/obj/")
 
-def compute_version_hash(filepath):
-  with open(filepath, "rb") as f:
-    # Normalize line endings on Windows
-    return int.from_bytes(hashlib.sha256(f.read().replace(b'\r', b'')).digest()[:4], 'little')
-
 USBPACKET_MAX_SIZE = 0x40
 
 class McuConfig(NamedTuple):

panda/python/spi.py

@@ -1,5 +1,6 @@
 import binascii
 import os
+import fcntl
 import math
 import time
 import struct
@@ -11,13 +12,6 @@ from .base import BaseHandle, BaseSTBootloaderHandle, TIMEOUT
 from .constants import McuType, MCU_TYPE_BY_IDCODE, USBPACKET_MAX_SIZE
 from .utils import logger
 
-# No fcntl on Windows
-try:
-  import fcntl
-except ImportError:
-  fcntl = None # type: ignore
-
-# No spidev on MacOS/Windows
 try:
   import spidev
 except ImportError: