From dd2214a78ccd932100d5987ef73196a7609645ca Mon Sep 17 00:00:00 2001 From: Daniel Koepping Date: Fri, 7 Aug 2026 02:35:17 -0700 Subject: [PATCH] chestnut updater (#38546) * chestnut: production flasher with ROM bootloader recovery * hardwared: flash chestnut firmware offroad * comments * single FW version * hardwared: fix chestnut flasher path * chestnut: update firmware to 1d368808 * hardwared: log the flasher output * comment * hardwared: flash in a thread * match panda flashing style * fix * chestnut: defer signals and harden the flasher * chestnut: detect bootloader mode and retry * comments * chestnut: name the config dir for its content * chestnut: name the ROM bootloader consistently * chestnut: clearer name for the device lookup * chestnut: spell out the low level names * chestnut: claim the interface, not the device --- openpilot/common/hardware/usb.py | 2 + openpilot/selfdrive/modeld/helpers.py | 5 +- .../hardware/chestnut/firmware_wrapped.bin | Bin 0 -> 9207 bytes openpilot/system/hardware/chestnut/flash.py | 556 ++++++++++++++++++ openpilot/system/hardware/hardwared.py | 45 +- 5 files changed, 605 insertions(+), 3 deletions(-) create mode 100644 openpilot/system/hardware/chestnut/firmware_wrapped.bin create mode 100755 openpilot/system/hardware/chestnut/flash.py diff --git a/openpilot/common/hardware/usb.py b/openpilot/common/hardware/usb.py index 0020a3783b..e06e77b31e 100644 --- a/openpilot/common/hardware/usb.py +++ b/openpilot/common/hardware/usb.py @@ -1,7 +1,9 @@ import os from pathlib import Path +CHESTNUT_FW_VERSION = "1d368808" CHESTNUT_USB_IDS = ((0xADD1, 0x0001), (0x3801, 0x0001)) +CHESTNUT_ROM_USB_IDS = ((0x174C, 0x2464), (0x174C, 0x2463)) USB_DEVICES_PATH = Path("/sys/bus/usb/devices") diff --git a/openpilot/selfdrive/modeld/helpers.py b/openpilot/selfdrive/modeld/helpers.py index f36259340c..37ab0b26d7 100644 --- a/openpilot/selfdrive/modeld/helpers.py +++ b/openpilot/selfdrive/modeld/helpers.py @@ -7,7 +7,7 @@ import tempfile from pathlib import Path from openpilot.common.file_chunker import get_manifest_path -from openpilot.common.hardware.usb import CHESTNUT_USB_IDS, USB_DEVICES_PATH +from openpilot.common.hardware.usb import CHESTNUT_FW_VERSION, CHESTNUT_USB_IDS, USB_DEVICES_PATH MODELS_DIR = Path(__file__).resolve().parent / 'models' TG_INPUT_DEVICES_PATH = MODELS_DIR / 'tg_input_devices.json' @@ -49,7 +49,8 @@ def usbgpu_present() -> bool: for d in USB_DEVICES_PATH.glob("*"): try: usb_id = (int((d / "idVendor").read_text(), 16), int((d / "idProduct").read_text(), 16)) - if usb_id in CHESTNUT_USB_IDS: + product = (d / "product").read_text().strip() + if usb_id in CHESTNUT_USB_IDS and product == f"custom {CHESTNUT_FW_VERSION}-CLEAN": return True except Exception: pass diff --git a/openpilot/system/hardware/chestnut/firmware_wrapped.bin b/openpilot/system/hardware/chestnut/firmware_wrapped.bin new file mode 100644 index 0000000000000000000000000000000000000000..a0911e06e23c4843d3ed07592ebaf5ea7a707669 GIT binary patch literal 9207 zcmb_ieQ*@VmG7DTnvnpB6(6=SX0Xnt`{y3&`0=d_!K~soiwRyob*auH;iF;Pc0qQ_{1eS#W?tW> zBGuU-;kL^Pa9n|>0iKKf@VXRQpf)&h+mHLz1{ZGo2^2}V?L#P@#BJxGScuy`AdD9T zMzy0odK*x^tUt~QX;DQW`O>1x$;zPS)LdGUHeXy|))s0{iqd}sEkSe8WJlju#xa|? z&_X~$Za1}?+bt!iE9HT*YT$$<6eS@oAcv1j?*!ETmrXIhG^6ahf_eU;5hW`*jEl^2 zl~f&2spcYG{W@q1TJ^4=O>4W2u*ugY?)~1t_i@btHYHNxhj)qM6}$$nXSmOi#*fuR zE>7gJu4cA{C|QW5u%O7t(AX3T8y0zEc_MQ06ef$jiI$rljd3()r?7b<@1U^+3j0Rn z%``TH!onhNrLmb5ri%O&8k-f3#nISo3WKp^)7YaF20HW8*kcq1{Y|5>ITQx^jMCZT z6bAZaX*malP1I*%G!~^#XEYYIel7|dn$TYog~1$b^jPLcV^KO=5RJvsatkSJqCOW< z*hGCMQy8o*L$|q@!Y1l7g~BH4^O9Z^nq!J|Yh_mc4d(hg@v2Xj#L9F#Hr4Z6mE4D&KoI5 zLVoIqaC6mg6?l7jM6m7EREyf@XIm0JR|g9jI3plVu^wTMBER_vjHrqeC1mG*7D}H; z-86~i7ZC%%x2A)dAz-={FAob3_T-x3OYKV?>CXs5KxGH(xs?Fc z18iV@x2DVPI|KRwcZs{q0m2+P$*krt zhe{Ho=BGusHTos^>b?b7U&O zV`BN=j*X7+$PdrSgXiRdbF$8Y8y%`e>V&5oo-Q(IY6AZ+<19L^4?wJCNdI6bVlAR6 zBvSBL2-Z6W3HvDrVXS7%ZT(cgp#viK5eX!SDACyHD==}>oN2DivzLIGKk3s$U?4$t0qeq&Gp@8XhQyvPqK?4?Q+9yV*olS8wG&e&5Wi&vCI#B?9W~;PB zI@Pf{+Hoe;zHt;sCYD73Z_v#c=dj2$)uUp|W~V(irIbo6siQImj zN*i(RwleAN zCuQ}}tj2If?&hSd3usm&o)x*?Nm(z@tVVb%ay^r>-lAEJXjkOQCuLnjv(mxupJ~+^ zk*~;oF=-oL(rrAcuY$-qCS^r5s}VDc+`dU!(`i;Cgci9ECS}$4)acGw99iwoZ>{!k z>D7LlvX*#yuE#0BxQaw}7VXdbX$ktC6$Q}FAZk0%x5V>wQ*_8bx*aY(eR-v@AZ18d zrQe%H?z7RUj$dvoo{1VNedAb5w?CDd+)f&x{YgFz(EcO}SVcMdov??FqJHc+T00C8 z6?Oz_hfY0JwSq)`5NpNmY`4YTfQ-hUB1|Jy0SNU7!bFgciO7`AqX;fW_9>rWAF{_C z61kt^;tH5QKCBb>j}i~-#1ftOol#_*hW&P}_Dp2V)AvOB z|1U^?iI97NK+i~=iSn@HtFg>kKS8rek_X>hBqXT#W-q%HV zA2Mu3Q2kzL6};b6;6-Blwqo*i7@jhC%2hU@vO&6%jXO}EdSx~4VEWYlv{A`tf#ppS za<&OXsu-RqNzzL=$Ho9Z$;y3?Fue(n5LyH82|YndP!0?6o1j!-EfOlPNF*kP8Cm+( z=I5V;$iKgCO}?XlzGqO$${D(xTP+D4dJay%2UV(ehZ^VqqDpd#&mwX4LNR-dD0$h0 zkqT)QeApBI&>-FLuzoWvwo~*xFXkXIH(Ol85^%-}0H2S5m%UfkHaz#?^JVDd3#Zm! zTvvud>pg7f1&^8BG^*7u=;>S!yhMr7!$w8hpl#Ihv`wOSgSJ+DE*k*dY-ijMy?NqV zq?1_tj2jhsT*W3{j6q^7!l+!)RC{9VMd`!9C3P4MUPa@z7%i5Z;iuGIyW=E*NZ>%6 zMVH!rPL1TOE^hd6;|Z=?<1^}HB7ylTv&(N2@RVKR}~pKcRi9Aw3eZAxDu5y<%kTT2X3ZU{bbRvl6&m^{a|A| zgEAE@eQ)EO&pD5pFJ5-Gm1MZtR;c=lV%hO$2fsDn|S5d>Q`am@rMn zpIC)NU``jNkfw#l^~Q6dv&J9Dx$%Jb74;}Of|_R_Nbk+wf5T*y_-ej9Oj^%AEf0~l zoqrnB1)omlS}Ey@;R{A5j_sSXi#Y?X3WDrJ9a)W`Xo%cZn4WWfzs)Y4U#*GF$4k%uv?*T2^S; zziG=t%YURT$M_7S5}c5aki!ny#LoE=KFQM8nB@^%W08k(jaAn2ag9yZHshLDd7uR0 zni%;(DO#^>>0YN5^f|urVAm{SmJzj7Nk*jryB^a?WMznU^=d@{QEbH)@&@%qWcrHY znk-W)@+G_NC)o8k%cm>Ne#SiDo7rTwS8QHVk)K+zC9R^sU6CQNw*Txb;+l^^+M-!n z+FO(s$}d_R+FZ2o{FV<3e!)7fqq&_vYY9^dMqPqR5hN|(j#PTPuW zcv;&7WJKmcs-se3U$CA7_Fs@PWB zF~4x2Cu{3NlBba$5+ZE8 zVbtMaJ3VGHh5@4L2c16lo!S$S(FcE|H%8R!D{lH)6#QE^YdbD(E?|$IV8*7f6Q;0> zNP;-bWB2*pMQkwx>IF5nI>B;rs|$WZ@UO$IGt|x}aqCR=(k|S(phvy96SpqxQbRky z5MX9t2uvBuVOiX22U}=e)JgJZeZGXIZ{kDwZrNg=wZd<*&x+Hj<|dVK61C2`-K3J_ z*qqx(!o-ruJ|r2I0k<4Ok`{18GybGb_Z{;Q)c+kdVD8V)w7ei@*@?T;5)mv46a&s6Q z$8zFL(%W@!eF_#~3HklR=!^8S-+#uReIM7`AWeigSdY*aZ-I|-!i@UecVbE65&FV4 zsnxW;!d1`1+ z(aJveSBbcu1Hc6U52X{z7$_5=jD>MP>d4SMQPf%B($ik-+-x0&6Kr4QOO<;oeMIWP zVf2pmxLFzSywvI0+vD-|dG?W3AC|&8R$5P5PMDO3o|n2kdwU}o*r+lvG}7UHNz3f? z?$xq-ygqGNpLd_OJe29p@-Fi(_a#8;bC2#P;^C~mw#Q_jT}d`DIL=2vx$e6*@@WRI z>Uu8X>#HNq4t6-kwQ|^frxZ%E9yL>Ch+uKeG`SRctlb_-Uv@=Eu0(|@riI0Dm`1toCK7MTAPY|yH zIq81)y#&%dr;b5n4$fd=P!6g=DHxUC2r@wwY!}-RZexu@VqO)lVPGk0S=k-3LLoTR zKDP|s^fO{3pQ+fj3(h{{IH;ujyp5e#)#ttgKdRVghue+Xy<}X=fS79^B@#ICP%_46iR^*DBa6>QWR(J3i{xSV-N&)3WK5vlCT1{H#vXXUmR%>u z7$Ot+0qi|ogJeA{h@4wC|5h;A6&T+ogLt13F(!cTMX_7j!7}cW9|G>d()%C9t}7s7 zZ0agnO<+Sau?zBRWWU>RYYI7$TbC$mf8j$PAU7wN6=7>{N269?geL8QcBx%$m)jGU z{-cF;{qY~QfP*%<@He0r1vp7^Ut%CdJ{Aeux%+ms!ET#@8)0|Ao}2|HVYmMr94a_g z2o>!%hj{*Nd%{)wjF0>y?)0S@nOV!0uXt*WCnwjt78M^ZDfR9D&Vg;)cJCf6aVSH> zBVls2(4YC}*R zhWZE)!Z{N2MRT{)@z?3@za$(FoR=IL|CjRtF@T)@k+1!l>V%LwBvo+g2Sbgq6X0TTI^Dq=>zy9rAgZ~~e}xB!7ts)&WMO05i> zlAMDdf9}3N^ADdZEp;$kr7i$Yy-w9s<-NgBto#1#MuKn}r_}Kpuhhk8F-l#m7TXOM z2lwX*gWnc&g0`J@2L7T0p=@RagWx}FV(g5=1QdRp!n3J}VNek9W`xY)a2VMv$5@cF zSy-b87=@V>JbQSYn;qjeBda?O+SO%7E6^Y_-k?r}KP@pQnJ6}$Uk5FuVnEsB!pt(O=ub-^BA|FeJ{VV|JB!!@cO~S=oRrz;m{k7^!*vjR;+NZ zSdz2ByCx5zUHLidUUeXcV{1OV_aA!eAaXqSD!hu3BWF`yo;Rn^n~V0uL>PtF-+1+q zD3E#%@S|O8$Y*kLvZKITNCeidd(HvXh=A8oxNf6&)7C;#-?*;hu=I_nfv@w5!x6liF}pmQHjz>O L`^@z2+MWLk;6m5R literal 0 HcmV?d00001 diff --git a/openpilot/system/hardware/chestnut/flash.py b/openpilot/system/hardware/chestnut/flash.py new file mode 100755 index 0000000000..cd9901b476 --- /dev/null +++ b/openpilot/system/hardware/chestnut/flash.py @@ -0,0 +1,556 @@ +#!/usr/bin/env python3 +"""chestnut (ASM2464) SPI flasher using data-USB EP0 control transfers.""" +import argparse +import ctypes +import errno +import fcntl +import glob +import hashlib +import os +import re +import signal +import struct +import sys +import time +import zlib +from pathlib import Path + +VID_PIDS = (("add1", "0001"), ("3801", "0001")) +ROM_VID_PIDS = (("174c", "2464"), ("174c", "2463")) +ROM_PRODUCT = "USB 3.2 PCIe TinyEnclosure" +FIRMWARE_PATH = Path(__file__).with_name("firmware_wrapped.bin") +CONFIG_DIR = "/data/chestnut_config" +PM_PATHS = ("/sys/bus/platform/devices/a600000.ssusb", "/sys/bus/usb/devices/usb4") +VBUS_PATH = "/sys/kernel/debug/regulator/smb2-vbus/enable" +IMAGE_OFFSET = 0x100 +SECTOR, PAGE = 4096, 128 +MAX_CODE_SIZE = 0x10000 +FLASH_BUDGET = 600.0 +USBDEVFS_CONTROL = 0xC0185500 +USBDEVFS_BULK = 0xC0185502 +USBDEVFS_SETINTERFACE = 0x80085504 +USBDEVFS_SETCONFIGURATION = 0x80045505 +USBDEVFS_CLAIMINTERFACE = 0x8004550F +USBDEVFS_RESET = 0x5514 +USBDEVFS_CLEAR_HALT = 0x80045515 + +_deadline = float("inf") + + +def check_budget(): + if time.monotonic() > _deadline: + raise TimeoutError(f"flash did not converge within {FLASH_BUDGET:g}s") + + +class Ctrl(ctypes.Structure): + _fields_ = [("request_type", ctypes.c_uint8), ("request", ctypes.c_uint8), + ("value", ctypes.c_uint16), ("index", ctypes.c_uint16), + ("length", ctypes.c_uint16), ("timeout", ctypes.c_uint32), + ("data", ctypes.c_void_p)] + + +class Bulk(ctypes.Structure): + _fields_ = [("ep", ctypes.c_uint), ("len", ctypes.c_uint), + ("timeout", ctypes.c_uint), ("data", ctypes.c_void_p)] + + +class RomFallback(Exception): + pass + + +def find_chestnut(): + found = [] + for d in glob.glob("/sys/bus/usb/devices/*"): + try: + vid_pid = (open(d + "/idVendor").read().strip(), open(d + "/idProduct").read().strip()) + if vid_pid in VID_PIDS + ROM_VID_PIDS: + found.append((d, vid_pid, open(d + "/product").read().strip())) + except OSError: + pass + if len(found) > 1: + raise RuntimeError(f"expected one chestnut, found {len(found)}") + return found[0] if found else (None, None, None) + + +def in_rom_bootloader(vid_pid, product): + # the ROM bootloader reports the config page strings, or its own when the config page is lost + return vid_pid in ROM_VID_PIDS or product == ROM_PRODUCT or (product or "").startswith("AS2462") + + +def disable_runtime_pm(path): + control = os.path.join(path, "power/control") + if not os.path.exists(control): + return + with open(control, "w") as f: + f.write("on\n") + if open(control).read().strip() != "on": + raise RuntimeError(f"could not disable USB runtime PM: {control}") + delay = os.path.join(path, "power/autosuspend_delay_ms") + if os.path.exists(delay): + with open(delay, "w") as f: + f.write("-1\n") + + +def unbind_drivers(path): + for interface in glob.glob(path + ":*"): + driver = interface + "/driver" + if os.path.islink(driver): + with open(os.path.realpath(driver) + "/unbind", "w") as f: + f.write(os.path.basename(interface)) + + +def open_device(path): + bus, dev = int(open(path + "/busnum").read()), int(open(path + "/devnum").read()) + return os.open(f"/dev/bus/usb/{bus:03d}/{dev:03d}", os.O_RDWR) + + +def claim_interface(path, setup=False): + # unbind usb-storage, which binds to the ROM bootloader + disable_runtime_pm(path) + unbind_drivers(path) + fd = open_device(path) + try: + if setup: + fcntl.ioctl(fd, USBDEVFS_SETCONFIGURATION, struct.pack("I", 1)) + fcntl.ioctl(fd, USBDEVFS_CLAIMINTERFACE, struct.pack("I", 0)) + if setup: + fcntl.ioctl(fd, USBDEVFS_SETINTERFACE, struct.pack("II", 0, 0)) + except OSError as e: + os.close(fd) + if e.errno == errno.EBUSY: + raise RuntimeError("chestnut is in use, stop modeld/GPU processes before flashing") from e + raise + return fd + + +class Flash: + def __init__(self): + self.fd = -1 + + def close(self): + if self.fd >= 0: + os.close(self.fd) + self.fd = -1 + + def connect(self, timeout=5.0): + self.close() + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + path, vid_pid, product = find_chestnut() + if in_rom_bootloader(vid_pid, product): + raise RomFallback("chestnut fell back to the ROM bootloader") + if path is not None: + self.fd = claim_interface(path) + return + time.sleep(0.1) + raise RuntimeError(f"chestnut did not enumerate within {timeout:g}s") + + def reg_write(self, addr, value): + fcntl.ioctl(self.fd, USBDEVFS_CONTROL, + Ctrl(0x40, 0xE5, addr & 0xFFFF, value & 0xFFFF, 0, 2000, None)) + + def reg_read(self, addr, length=1): + buf = (ctypes.c_ubyte * length)() + fcntl.ioctl(self.fd, USBDEVFS_CONTROL, + Ctrl(0xC0, 0xE4, addr & 0xFFFF, 0, length, 2000, ctypes.cast(buf, ctypes.c_void_p))) + return bytes(buf) + + def write_buffer(self, data): + for i, value in enumerate(data): + self.reg_write(0x7000 + i, value) + + def wait_controller(self, timeout=2.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if not self.reg_read(0xC8A9)[0] & 1: + return + raise TimeoutError("flash controller timeout") + + def transaction(self, command, addr=0, length=0, addr_len=0x07, mode=0): + for reg, value in ((0xC8AD, mode), (0xC8AE, 0), (0xC8AF, 0), (0xC8AA, command), (0xC8AC, addr_len), + (0xC8A1, addr), (0xC8A2, addr >> 8), (0xC8AB, addr >> 16), (0xC8A3, length >> 8), (0xC8A4, length)): + self.reg_write(reg, value & 0xFF) + self.reg_write(0xC8A9, 1) + self.wait_controller() + for _ in range(4): + self.reg_write(0xC8AD, 0) + + def write_enable(self): + for reg, value in ((0xC8AD, 0), (0xC8AA, 0x06), (0xC8AC, 0x04), (0xC8A3, 0), (0xC8A4, 0), (0xC8A9, 1)): + self.reg_write(reg, value) + self.wait_controller() + + def status(self): + self.transaction(0x05, length=1, addr_len=0x04) + return self.reg_read(0x7000)[0] + + def wait_write_done(self, timeout=10.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if not self.status() & 1: + return + time.sleep(0.005) + raise TimeoutError("SPI flash WIP timeout") + + def init(self): + self.reg_write(0xCC33, 0x04) + self.reg_write(0xCA81, self.reg_read(0xCA81)[0] | 1) + self.reg_write(0xC805, 0x02) + self.reg_write(0xC8A6, 0x04) + for _ in range(5): + self.write_enable() + self.write_buffer(bytes(4)) + self.transaction(0x01, length=1, addr_len=0x04, mode=1) + time.sleep(0.01) + if not self.status() & 0x1C: + return + raise RuntimeError("could not clear SPI block protection") + + def read(self, addr, length): + out = bytearray() + while len(out) < length: + n = min(4096, length - len(out)) + self.transaction(0x03, addr + len(out), max(4096, n)) + for off in range(0, n, 255): + out += self.reg_read(0x7000 + off, min(255, n - off)) + return bytes(out) + + def erase_sector(self, addr): + self.write_enable() + self.transaction(0x20, addr) + self.wait_write_done() + + def program(self, addr, data): + self.write_buffer(data + bytes((-len(data)) % 4)) + self.write_enable() + self.transaction(0x02, addr, len(data), mode=1) + self.wait_write_done() + + +def validate_image(data): + if len(data) < 10: + raise ValueError("wrapped firmware is too short") + body_len = int.from_bytes(data[:4], "little") + if body_len > MAX_CODE_SIZE: + raise ValueError(f"wrapped firmware body exceeds {MAX_CODE_SIZE} bytes") + if len(data) != body_len + 10 or data[4 + body_len] != 0xA5: + raise ValueError("invalid wrapped firmware length or magic") + body = data[4:4 + body_len] + if data[5 + body_len] != sum(body) & 0xFF: + raise ValueError("invalid wrapped firmware checksum") + if data[6 + body_len:] != zlib.crc32(body).to_bytes(4, "little"): + raise ValueError("invalid wrapped firmware CRC") + + +def image_product(image): + match = re.search(rb"custom [0-9a-f]{8}-CLEAN", image) + if match is None: + raise ValueError("no product string in wrapped firmware") + return match.group().decode() + + +def reconnect(flash): + attempt = 0 + while True: + attempt += 1 + check_budget() + try: + flash.connect() + flash.init() + return + except (OSError, TimeoutError, RuntimeError) as e: + print(f"waiting for chestnut (attempt {attempt}): {e}", flush=True) + time.sleep(1) + + +def with_retries(flash, label, operation): + # on any transfer error, reconnect and restart the operation + attempt = 0 + while True: + attempt += 1 + try: + return operation() + except (OSError, TimeoutError, RuntimeError) as e: + check_budget() + print(f"{label} attempt {attempt}: {e}", flush=True) + reconnect(flash) + + +def stable_read(flash, addr, length, count=2): + def read(): + reads = [flash.read(addr, length) for _ in range(count)] + if any(x != reads[0] for x in reads[1:]): + raise RuntimeError(f"unstable flash read at 0x{addr:05x}") + return reads[0] + return with_retries(flash, f"read 0x{addr:05x}", read) + + +def program_sector(flash, addr, target): + def program(): + flash.erase_sector(addr) + if flash.read(addr, SECTOR) != bytes([0xFF]) * SECTOR: + raise RuntimeError("sector erase verification failed") + for off in range(0, SECTOR, PAGE): + chunk = target[off:off + PAGE] + if chunk != bytes([0xFF]) * len(chunk): + flash.program(addr + off, chunk) + if flash.read(addr + off, len(chunk)) != chunk: + raise RuntimeError(f"page verify failed at 0x{addr + off:05x}") + if flash.read(addr, SECTOR) != target: + raise RuntimeError("sector verification failed") + with_retries(flash, f"sector 0x{addr:05x}", program) + + +def config_path(): + return os.path.join(CONFIG_DIR, f"{os.uname().nodename}.bin") + + +def saved_config(path, data): + os.makedirs(os.path.dirname(path), exist_ok=True) + try: + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError as e: + backup = open(path, "rb").read() + if len(backup) != 0x100: + raise RuntimeError(f"invalid config backup: {path}") from e + if backup != data: + print(f"restoring config from {path}", flush=True) + return backup + with os.fdopen(fd, "wb") as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + return data + + +def rom_write(image, config): + # the ROM bootloader implements only the BOT protocol, and requires a port reset before bulk transfers + path, _, _ = find_chestnut() + if path is None: + raise RuntimeError("chestnut disappeared before recovery") + unbind_drivers(path) + fd = open_device(path) + try: + fcntl.ioctl(fd, USBDEVFS_RESET) + finally: + os.close(fd) + time.sleep(3) + path, _, _ = find_chestnut() + if path is None: + raise RuntimeError("chestnut did not re-enumerate after reset") + fd = claim_interface(path, setup=True) + for ep in (0x02, 0x81): + fcntl.ioctl(fd, USBDEVFS_CLEAR_HALT, struct.pack("I", ep)) + tag = 0 + + def bulk(ep, payload, timeout): + buf = ctypes.create_string_buffer(bytes(payload), len(payload)) + fcntl.ioctl(fd, USBDEVFS_BULK, Bulk(ep, len(payload), timeout, ctypes.cast(buf, ctypes.c_void_p))) + return buf.raw + + def cmd(cdb, data=b"", timeout=30000): + nonlocal tag + tag += 1 + bulk(0x02, struct.pack("BBB12x", 0xE1, 0x50, 0), config[:0x80]) + cmd(struct.pack(">BBB12x", 0xE1, 0x50, 1), config[0x80:]) + cmd(struct.pack(">BBI", 0xE3, 0x50, min(len(image), 0xFF00)), image[:0xFF00]) + if len(image) > 0xFF00: + cmd(struct.pack(">BBI", 0xE3, 0xD0, len(image) - 0xFF00), image[0xFF00:]) + cmd(struct.pack(">BB13x", 0xE8, 0x51)) + finally: + os.close(fd) + print("recovery flash done", flush=True) + + +def vbus_write(value): + try: + with open(VBUS_PATH, "w") as f: + f.write(value + "\n") + except OSError: + pass + + +def vbus_cycle(): + if os.path.exists(VBUS_PATH): + vbus_write("0") + time.sleep(2) + vbus_write("1") + time.sleep(5) + + +def activate(expected_product): + if not os.path.exists(VBUS_PATH): + print("no VBUS control, firmware activates on the next chestnut power cycle", flush=True) + return + print("power-cycling chestnut VBUS", flush=True) + vbus_write("0") + disconnected = False + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + path, _, _ = find_chestnut() + if path is None: + disconnected = True + break + time.sleep(0.2) + time.sleep(1) + vbus_write("1") + if not disconnected: + print("chestnut stayed powered, firmware activates on its next power cycle", flush=True) + return + deadline = time.monotonic() + 15.0 + while time.monotonic() < deadline: + _, _, product = find_chestnut() + if product is not None: + if product == expected_product: + print(f"activated {expected_product}", flush=True) + else: + print(f"chestnut re-enumerated with {product!r}, firmware activates on its next power cycle", flush=True) + return + time.sleep(0.2) + print("chestnut did not re-enumerate, firmware activates on its next power cycle", flush=True) + + +def defer_signal(signum, _frame): + # writing from a handler must not reenter a print already in progress + os.write(1, f"signal {signum} deferred until the chestnut is powered back up\n".encode()) + + +def flash_chestnut(expected_version=None, force=False): + global _deadline + + image = FIRMWARE_PATH.read_bytes() + validate_image(image) + expected_product = image_product(image) + if expected_version is not None and expected_product != f"custom {expected_version}-CLEAN": + raise RuntimeError(f"bundled firmware is {expected_product!r}, expected version {expected_version}") + + path, vid_pid, product = find_chestnut() + if path is None: + print("no chestnut connected", flush=True) + return + if product == expected_product and not force: + print(f"chestnut firmware is up to date ({expected_product})", flush=True) + return + + _deadline = time.monotonic() + FLASH_BUDGET + for pm_path in PM_PATHS: + disable_runtime_pm(pm_path) + + previous = {sig: signal.signal(sig, defer_signal) for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP)} + try: + if in_rom_bootloader(vid_pid, product): + if not recover_from_rom(image, expected_product): + return + # firmware is back, verify it against the bundled image + force, product = True, None + write_image(image, expected_product, product, force) + finally: + for sig, handler in previous.items(): + signal.signal(sig, handler) + + +def recover_from_rom(image, expected_product): + # returns whether the chestnut came back on custom firmware + backup = config_path() + if not os.path.isfile(backup): + raise RuntimeError(f"cannot recover from the ROM bootloader without a config backup at {backup}") + config = open(backup, "rb").read() + if len(config) != 0x100: + raise RuntimeError(f"invalid config backup: {backup}") + + committed = False + while True: + check_budget() + path, vid_pid, product = find_chestnut() + if path is None: + if committed: + print("chestnut is offline, recovered firmware boots on its next power cycle", flush=True) + return False + vbus_cycle() + continue + if not in_rom_bootloader(vid_pid, product): + return True + if committed: + print("chestnut stayed powered, recovered firmware boots on its next power cycle", flush=True) + return False + try: + rom_write(image, config) + committed = True + except (OSError, TimeoutError, RuntimeError) as e: + print(f"ROM recovery failed, retrying: {e}", flush=True) + vbus_cycle() + continue + activate(expected_product) + + +def write_image(image, expected_product, product, force): + if force: + print(f"forced reflash of {expected_product}", flush=True) + else: + print(f"chestnut firmware mismatch: {product!r}; expected {expected_product!r}", flush=True) + + flash = Flash() + try: + reconnect(flash) + config = stable_read(flash, 0, 0x100, 3) + config = saved_config(config_path(), config) + image_end = IMAGE_OFFSET + len(image) + first_sector = IMAGE_OFFSET & ~(SECTOR - 1) + span = (image_end + SECTOR - 1) & ~(SECTOR - 1) + current = stable_read(flash, first_sector, span - first_sector) + target = bytearray(current) + target[:len(config)] = config + target[IMAGE_OFFSET - first_sector:image_end - first_sector] = image + target = bytes(target) + print(f"target {len(image)} bytes at 0x{IMAGE_OFFSET:05x}, sha256={hashlib.sha256(image).hexdigest()}", flush=True) + + for addr in range(first_sector, span, SECTOR): + off = addr - first_sector + wanted = target[off:off + SECTOR] + if current[off:off + SECTOR] == wanted: + print(f"sector 0x{addr:05x}: unchanged", flush=True) + else: + print(f"sector 0x{addr:05x}: programming", flush=True) + program_sector(flash, addr, wanted) + + verified = stable_read(flash, first_sector, span - first_sector, 3) + if verified != target: + raise RuntimeError("final full-image verification failed") + print(f"verified sha256={hashlib.sha256(verified).hexdigest()}", flush=True) + finally: + flash.close() + + activate(expected_product) + + +def main(): + parser = argparse.ArgumentParser(description="check and flash the bundled chestnut firmware") + parser.add_argument("version", nargs="?", help="expected firmware version hash") + parser.add_argument("--force", action="store_true", help="reflash even when the version matches") + args = parser.parse_args() + if os.geteuid() != 0: + raise RuntimeError("flash.py must run as root") + flash_chestnut(expected_version=args.version, force=args.force) + + +if __name__ == "__main__": + try: + main() + except Exception as e: + print(f"FAIL: {type(e).__name__}: {e}", file=sys.stderr) + sys.exit(1) diff --git a/openpilot/system/hardware/hardwared.py b/openpilot/system/hardware/hardwared.py index 69f118d337..10ee0aee3b 100755 --- a/openpilot/system/hardware/hardwared.py +++ b/openpilot/system/hardware/hardwared.py @@ -3,6 +3,8 @@ import fcntl import os import queue import struct +import subprocess +import sys import threading import time from collections import OrderedDict, namedtuple @@ -16,7 +18,8 @@ from openpilot.common.params import Params from openpilot.common.realtime import DT_HW from openpilot.selfdrive.selfdrived.alertmanager import set_offroad_alert from openpilot.common.hardware import HARDWARE, TICI, PC -from openpilot.common.hardware.usb import get_usb_state, get_usb_topology, set_usb_state +from openpilot.common.basedir import BASEDIR +from openpilot.common.hardware.usb import CHESTNUT_FW_VERSION, CHESTNUT_ROM_USB_IDS, CHESTNUT_USB_IDS, get_usb_state, get_usb_topology, set_usb_state from openpilot.common.linux import LinuxSystemStats from openpilot.system.loggerd.config import get_available_percent from openpilot.common.swaglog import cloudlog @@ -34,6 +37,44 @@ DISCONNECT_TIMEOUT = 5. # wait 5 seconds before going offroad after disconnect PANDA_STATES_TIMEOUT = round(1000 / SERVICE_LIST['pandaStates'].frequency * 1.5) # 1.5x the expected pandaState frequency ONROAD_CYCLE_TIME = 1 # seconds to wait offroad after requesting an onroad cycle +class Chestnut: + # flash offroad, modeld ignores chestnut until the product string matches + MAX_ATTEMPTS = 3 + RETRY_INTERVAL = 20. + + def __init__(self): + self.thread: threading.Thread | None = None + self.attempts = 0 + self.last_attempt = 0. + self.flashed = False + + def flash(self) -> None: + ret = subprocess.run(["sudo", sys.executable, os.path.join(BASEDIR, "openpilot/system/hardware/chestnut/flash.py"), CHESTNUT_FW_VERSION], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False) + cloudlog.event("chestnut flash done", returncode=ret.returncode, output=ret.stdout[-1000:], error=ret.returncode != 0) + self.flashed = ret.returncode == 0 + + def update(self, offroad: bool, usb_state: list[dict]) -> None: + mismatch = any((d["vendorId"], d["productId"]) in CHESTNUT_USB_IDS + CHESTNUT_ROM_USB_IDS and + d["product"] != f"custom {CHESTNUT_FW_VERSION}-CLEAN" for d in usb_state) + if not mismatch: + self.flashed = False + return + + if not offroad or self.flashed or self.attempts >= self.MAX_ATTEMPTS: + return + if self.thread is not None and self.thread.is_alive(): + return + if time.monotonic() - self.last_attempt < self.RETRY_INTERVAL: + return + + self.attempts += 1 + self.last_attempt = time.monotonic() + cloudlog.warning(f"chestnut firmware out of date, flashing (attempt {self.attempts})") + self.thread = threading.Thread(target=self.flash, daemon=True) + self.thread.start() + + ThermalBand = namedtuple("ThermalBand", ['min_temp', 'max_temp']) HardwareState = namedtuple("HardwareState", ['network_type', 'network_info', 'network_strength', 'network_stats', 'network_metered', 'modem_temps', 'usb_state']) @@ -195,6 +236,7 @@ def hardware_thread(end_event, hw_queue) -> None: thermal_config = HARDWARE.get_thermal_config() fan_controller = FanController(int(1./DT_HW)) + chestnut = Chestnut() while not end_event.is_set(): sm.update(PANDA_STATES_TIMEOUT) @@ -255,6 +297,7 @@ def hardware_thread(end_event, hw_queue) -> None: msg.deviceState.screenBrightnessPercent = HARDWARE.get_screen_brightness() set_usb_state(msg.deviceState, last_hw_state.usb_state) + chestnut.update(started_ts is None, last_hw_state.usb_state) # this subset is only used for offroad temp_sources = [