IQ.Pilot Prebuilt Release @ da36e21

_vendor_runtime/panda/board/body/__init__.py

@@ -0,0 +1,42 @@
+# python helpers for the body panda
+import struct
+
+from panda import Panda
+
+class PandaBody(Panda):
+
+  MOTOR_LEFT: int = 1
+  MOTOR_RIGHT: int = 2
+
+  # ****************** Motor Control *****************
+  @staticmethod
+  def _ensure_valid_motor(motor: int) -> None:
+    if motor not in (PandaBody.MOTOR_LEFT, PandaBody.MOTOR_RIGHT):
+      raise ValueError("motor must be MOTOR_LEFT or MOTOR_RIGHT")
+
+  def motor_set_speed(self, motor: int, speed: int) -> None:
+    self._ensure_valid_motor(motor)
+    assert -100 <= speed <= 100, "speed must be between -100 and 100"
+    speed_param = speed if speed >= 0 else (256 + speed)
+    self._handle.controlWrite(Panda.REQUEST_OUT, 0xe0, motor, speed_param, b'')
+
+  def motor_set_target_rpm(self, motor: int, rpm: float) -> None:
+    self._ensure_valid_motor(motor)
+    target_deci_rpm = int(round(rpm * 10.0))
+    assert -32768 <= target_deci_rpm <= 32767, "target rpm out of range (-3276.8 to 3276.7)"
+    target_param = target_deci_rpm if target_deci_rpm >= 0 else (target_deci_rpm + (1 << 16))
+    self._handle.controlWrite(Panda.REQUEST_OUT, 0xe4, motor, target_param, b'')
+
+  def motor_stop(self, motor: int) -> None:
+    self._ensure_valid_motor(motor)
+    self._handle.controlWrite(Panda.REQUEST_OUT, 0xe1, motor, 0, b'')
+
+  def motor_get_encoder_state(self, motor: int) -> tuple[int, float]:
+    self._ensure_valid_motor(motor)
+    dat = self._handle.controlRead(Panda.REQUEST_IN, 0xe2, motor, 0, 8)
+    position, rpm_milli = struct.unpack("<ii", dat)
+    return position, rpm_milli / 1000.0
+
+  def motor_reset_encoder(self, motor: int) -> None:
+    self._ensure_valid_motor(motor)
+    self._handle.controlWrite(Panda.REQUEST_OUT, 0xe3, motor, 0, b'')

_vendor_runtime/panda/board/body/flash.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+import argparse
+import os
+import subprocess
+
+from panda import Panda
+
+BODY_DIR = os.path.dirname(os.path.realpath(__file__))
+BOARD_DIR = os.path.abspath(os.path.join(BODY_DIR, ".."))
+REPO_ROOT = os.path.abspath(os.path.join(BOARD_DIR, ".."))
+DEFAULT_FIRMWARE = os.path.join(BOARD_DIR, "obj", "body_h7.bin.signed")
+
+
+def build_body() -> None:
+  subprocess.check_call(
+    f"scons -C {REPO_ROOT} -j$(nproc) board/obj/body_h7.bin.signed",
+    shell=True,
+  )
+
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser()
+  parser.add_argument("firmware", nargs="?", help="Optional path to firmware binary to flash")
+  parser.add_argument("--all", action="store_true", help="Flash all Panda devices")
+  parser.add_argument(
+    "--wait-usb",
+    action="store_true",
+    help="Wait for the panda to reconnect over USB after flashing (defaults to skipping reconnect).",
+  )
+  args = parser.parse_args()
+
+  firmware_path = os.path.abspath(args.firmware) if args.firmware is not None else DEFAULT_FIRMWARE
+
+  build_body()
+
+  if not os.path.isfile(firmware_path):
+    parser.error(f"firmware file not found: {firmware_path}")
+
+  if args.all:
+    serials = Panda.list()
+    print(f"found {len(serials)} panda(s) - {serials}")
+  else:
+    serials = [None]
+
+  for s in serials:
+    with Panda(serial=s) as p:
+      print("flashing", p.get_usb_serial())
+      p.flash(firmware_path, reconnect=args.wait_usb)
+  exit(1 if len(serials) == 0 else 0)

_vendor_runtime/panda/board/flash.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+import os
+import subprocess
+import argparse
+
+from panda import Panda
+
+board_path = os.path.dirname(os.path.realpath(__file__))
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser()
+  parser.add_argument("--all", action="store_true", help="Recover all Panda devices")
+  args = parser.parse_args()
+
+  subprocess.check_call(f"scons -C {board_path}/.. -j$(nproc) {board_path}", shell=True)
+
+  if args.all:
+    serials = Panda.list()
+    print(f"found {len(serials)} panda(s) - {serials}")
+  else:
+    serials = [None]
+
+  for s in serials:
+    with Panda(serial=s) as p:
+      print("flashing", p.get_usb_serial())
+      p.flash()
+  exit(1 if len(serials) == 0 else 0)

_vendor_runtime/panda/board/jungle/__init__.py

@@ -0,0 +1,159 @@
+# python library to interface with panda
+import os
+import struct
+from functools import wraps
+
+from panda import Panda, PandaDFU
+from panda.python.constants import McuType
+
+BASEDIR = os.path.dirname(os.path.realpath(__file__))
+FW_PATH = os.path.join(BASEDIR, "../obj/")
+
+
+def ensure_jungle_health_packet_version(fn):
+  @wraps(fn)
+  def wrapper(self, *args, **kwargs):
+    if self.health_version != self.HEALTH_PACKET_VERSION:
+      raise RuntimeError(f"Jungle firmware ({self.health_version}) doesn't match the \
+                           library's health packet version ({self.HEALTH_PACKET_VERSION}). \
+                           Reflash jungle.")
+    return fn(self, *args, **kwargs)
+  return wrapper
+
+
+class PandaJungleDFU(PandaDFU):
+  def recover(self):
+    fn = os.path.join(FW_PATH, self._mcu_type.config.bootstub_fn.replace("panda", "panda_jungle"))
+    with open(fn, "rb") as f:
+      code = f.read()
+    self.program_bootstub(code)
+    self.reset()
+
+
+class PandaJungle(Panda):
+  USB_PIDS = (0xddef, 0xddcf)
+
+  HW_TYPE_UNKNOWN = b'\x00'
+  HW_TYPE_V2 = b'\x02'
+
+  H7_DEVICES = [HW_TYPE_V2, ]
+  SUPPORTED_DEVICES = H7_DEVICES
+
+  HEALTH_PACKET_VERSION = 1
+  HEALTH_STRUCT = struct.Struct("<IffffffHHHHHHHHHHHH")
+
+  HARNESS_ORIENTATION_NONE = 0
+  HARNESS_ORIENTATION_1 = 1
+  HARNESS_ORIENTATION_2 = 2
+
+  @classmethod
+  def spi_connect(cls, serial, ignore_version=False):
+    return None, None, None, None
+
+  def flash(self, fn=None, code=None, reconnect=True):
+    if not fn:
+      fn = os.path.join(FW_PATH, self._mcu_type.config.app_fn.replace("panda", "panda_jungle"))
+    super().flash(fn=fn, code=code, reconnect=reconnect)
+
+  def recover(self, timeout: int | None = 60, reset: bool = True) -> bool:
+    dfu_serial = self.get_dfu_serial()
+
+    if reset:
+      self.reset(enter_bootstub=True)
+      self.reset(enter_bootloader=True)
+
+    if not self.wait_for_dfu(dfu_serial, timeout=timeout):
+      return False
+
+    dfu = PandaJungleDFU(dfu_serial)
+    dfu.recover()
+
+    # reflash after recover
+    self.connect(True, True)
+    self.flash()
+    return True
+
+  def get_mcu_type(self) -> McuType:
+    hw_type = self.get_type()
+    if hw_type in PandaJungle.H7_DEVICES:
+      return McuType.H7
+    raise ValueError(f"unknown HW type: {hw_type}")
+
+  def up_to_date(self, fn=None) -> bool:
+    if fn is None:
+      fn = os.path.join(FW_PATH, self.get_mcu_type().config.app_fn.replace("panda", "panda_jungle"))
+    return super().up_to_date(fn=fn)
+
+  # ******************* health *******************
+
+  @ensure_jungle_health_packet_version
+  def health(self):
+    dat = self._handle.controlRead(PandaJungle.REQUEST_IN, 0xd2, 0, 0, self.HEALTH_STRUCT.size)
+    a = self.HEALTH_STRUCT.unpack(dat)
+    return {
+      "uptime": a[0],
+      "ch1_power": a[1],
+      "ch2_power": a[2],
+      "ch3_power": a[3],
+      "ch4_power": a[4],
+      "ch5_power": a[5],
+      "ch6_power": a[6],
+      "ch1_sbu1_voltage": a[7] / 1000.0,
+      "ch1_sbu2_voltage": a[8] / 1000.0,
+      "ch2_sbu1_voltage": a[9] / 1000.0,
+      "ch2_sbu2_voltage": a[10] / 1000.0,
+      "ch3_sbu1_voltage": a[11] / 1000.0,
+      "ch3_sbu2_voltage": a[12] / 1000.0,
+      "ch4_sbu1_voltage": a[13] / 1000.0,
+      "ch4_sbu2_voltage": a[14] / 1000.0,
+      "ch5_sbu1_voltage": a[15] / 1000.0,
+      "ch5_sbu2_voltage": a[16] / 1000.0,
+      "ch6_sbu1_voltage": a[17] / 1000.0,
+      "ch6_sbu2_voltage": a[18] / 1000.0,
+    }
+
+  # ******************* control *******************
+
+  # Returns tuple with health packet version and CAN packet/USB packet version
+  def get_packets_versions(self):
+    dat = self._handle.controlRead(PandaJungle.REQUEST_IN, 0xdd, 0, 0, 3)
+    if dat and len(dat) == 3:
+      a = struct.unpack("BBB", dat)
+      return (a[0], a[1], a[2])
+    return (-1, -1, -1)
+
+  # ******************* jungle stuff *******************
+
+  def set_panda_power(self, enabled):
+    self._handle.controlWrite(PandaJungle.REQUEST_OUT, 0xa0, int(enabled), 0, b'')
+
+  def set_panda_individual_power(self, port, enabled):
+    self._handle.controlWrite(PandaJungle.REQUEST_OUT, 0xa3, int(port), int(enabled), b'')
+
+  def set_harness_orientation(self, mode):
+    self._handle.controlWrite(PandaJungle.REQUEST_OUT, 0xa1, int(mode), 0, b'')
+
+  def set_ignition(self, enabled):
+    self._handle.controlWrite(PandaJungle.REQUEST_OUT, 0xa2, int(enabled), 0, b'')
+
+  def set_can_silent(self, silent):
+    self._handle.controlWrite(PandaJungle.REQUEST_OUT, 0xf5, int(silent), 0, b'')
+
+  def set_generated_can(self, enabled):
+    self._handle.controlWrite(PandaJungle.REQUEST_OUT, 0xa4, int(enabled), 0, b'')
+
+  # ******************* serial *******************
+
+  def debug_read(self):
+    ret = []
+    while 1:
+      lret = bytes(self._handle.controlRead(PandaJungle.REQUEST_IN, 0xe0, 0, 0, 0x40))
+      if len(lret) == 0:
+        break
+      ret.append(lret)
+    return b''.join(ret)
+
+  # ******************* header pins *******************
+
+  def set_header_pin(self, pin_num, enabled):
+    self._handle.controlWrite(Panda.REQUEST_OUT, 0xf7, int(pin_num), int(enabled), b'')

_vendor_runtime/panda/board/jungle/flash.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+import os
+import subprocess
+import argparse
+
+from panda import PandaJungle
+
+board_path = os.path.dirname(os.path.realpath(__file__))
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser()
+  parser.add_argument("--all", action="store_true", help="Recover all panda jungle devices")
+  args = parser.parse_args()
+
+  subprocess.check_call(f"scons -C {board_path}/.. -u -j$(nproc) .", shell=True)
+
+  if args.all:
+    serials = PandaJungle.list()
+    print(f"found {len(serials)} panda jungles(s) - {serials}")
+  else:
+    serials = [None]
+
+  for s in serials:
+    with PandaJungle(serial=s) as p:
+      print("flashing", p.get_usb_serial())
+      p.flash()
+  exit(1 if len(serials) == 0 else 0)

_vendor_runtime/panda/board/jungle/recover.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+import os
+import time
+import subprocess
+import argparse
+
+from panda import PandaJungle, PandaJungleDFU
+
+board_path = os.path.dirname(os.path.realpath(__file__))
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser()
+  parser.add_argument("--all", action="store_true", help="Recover all panda jungle devices")
+  args = parser.parse_args()
+
+  subprocess.check_call(f"scons -C {board_path}/.. -u -j$(nproc) .", shell=True)
+
+  serials = PandaJungle.list() if args.all else [None]
+  for s in serials:
+    with PandaJungle(serial=s) as p:
+      print(f"putting {p.get_usb_serial()} in DFU mode")
+      p.reset(enter_bootstub=True)
+      p.reset(enter_bootloader=True)
+
+  # wait for reset panda jungles to come back up
+  time.sleep(1)
+
+  dfu_serials = PandaJungleDFU.list()
+  print(f"found {len(dfu_serials)} panda jungle(s) in DFU - {dfu_serials}")
+  for s in dfu_serials:
+    print("flashing", s)
+    PandaJungleDFU(s).recover()
+  exit(1 if len(dfu_serials) == 0 else 0)

_vendor_runtime/panda/board/recover.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+import os
+import time
+import subprocess
+
+from panda import Panda, PandaDFU
+
+board_path = os.path.dirname(os.path.realpath(__file__))
+
+if __name__ == "__main__":
+  subprocess.check_call(f"scons -C {board_path}/.. -j$(nproc) {board_path}", shell=True)
+
+  for s in Panda.list():
+    with Panda(serial=s) as p:
+      print(f"putting {p.get_usb_serial()} in DFU mode")
+      p.reset(enter_bootstub=True)
+      p.reset(enter_bootloader=True)
+
+  # wait for reset pandas to come back up
+  time.sleep(1)
+
+  dfu_serials = PandaDFU.list()
+  print(f"found {len(dfu_serials)} panda(s) in DFU - {dfu_serials}")
+  for s in dfu_serials:
+    print("flashing", s)
+    PandaDFU(s).recover()
+  exit(1 if len(dfu_serials) == 0 else 0)