mirror of
https://github.com/sunnypilot/sunnypilot.git
synced 2026-08-21 22:13:45 +08:00
@@ -1,7 +1,6 @@
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
import sys, subprocess
|
||||
import time
|
||||
from SCons.Script import Action, Value
|
||||
from openpilot.common.file_chunker import chunk_file, get_chunk_targets, get_existing_chunks
|
||||
@@ -88,18 +87,13 @@ for usbgpu in [False, True] if USBGPU else [False]:
|
||||
onnx_sizes_sum = sum(os.path.getsize(f) for f in driving_onnx_deps)
|
||||
chunk_targets = get_chunk_targets(target_pkl_path, estimate_pickle_max_size(onnx_sizes_sum))
|
||||
def do_compile(target, source, env, command=cmd, pkl=target_pkl_path, chunks=chunk_targets):
|
||||
from openpilot.system.hardware.chestnut.flash import link_up
|
||||
# chestnut can enumerate before its PCIe link is up due to varying 12V power behavior across cars
|
||||
probe_cmd = [sys.executable, '-c', 'from openpilot.selfdrive.modeld.compile_modeld import Device; Device[Device.DEFAULT]']
|
||||
probe_env = {**env['ENV'], 'DEV': 'USB+AMD'}
|
||||
ready = False
|
||||
for attempt in range(3):
|
||||
if attempt: time.sleep(1)
|
||||
try:
|
||||
ready = subprocess.run(probe_cmd, env=probe_env, capture_output=True, timeout=10).returncode == 0
|
||||
except subprocess.TimeoutExpired:
|
||||
pass
|
||||
if ready: break
|
||||
if not ready:
|
||||
for _ in range(10):
|
||||
if link_up():
|
||||
break
|
||||
time.sleep(1)
|
||||
else:
|
||||
print("Chestnut not ready, skipping big model build")
|
||||
return
|
||||
if ret := env.Execute(command):
|
||||
|
||||
@@ -104,6 +104,26 @@ def open_device(path):
|
||||
return os.open(f"/dev/bus/usb/{bus:03d}/{dev:03d}", os.O_RDWR)
|
||||
|
||||
|
||||
def link_up() -> bool:
|
||||
# asm enumerates on USB-C alone, gpu is only usable once pcie link is up
|
||||
try:
|
||||
path, _, _ = find_chestnut()
|
||||
if path is None:
|
||||
return False
|
||||
fd = open_device(path)
|
||||
except (OSError, RuntimeError):
|
||||
return False
|
||||
try:
|
||||
fcntl.ioctl(fd, USBDEVFS_CONTROL, Ctrl(0x40, 0xF3, 1, 0, 0, 2000, None))
|
||||
buf = (ctypes.c_ubyte * 1)()
|
||||
fcntl.ioctl(fd, USBDEVFS_CONTROL, Ctrl(0xC0, 0xE4, 0xB450, 0, 1, 1000, ctypes.cast(buf, ctypes.c_void_p)))
|
||||
return buf[0] == 0x78 # LTSSM L0
|
||||
except OSError:
|
||||
return False
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
|
||||
def claim_interface(path, setup=False):
|
||||
# unbind usb-storage, which binds to the ROM bootloader
|
||||
disable_runtime_pm(path)
|
||||
|
||||
Reference in New Issue
Block a user