add chestnut offroad alerts (#38706)

* system: add chestnut offroad alerts

* system: refine chestnut offroad alerts

* system: refine chestnut power alerts

* system: confirm chestnut power recovery from PCIe

* system: detect missing chestnut power from INA voltage

openpilot/common/hardware/usb.py

@@ -4,11 +4,17 @@ from pathlib import Path
 CHESTNUT_FW_VERSION = "ed4e39b7"
 CHESTNUT_USB_IDS = ((0xADD1, 0x0001), (0x3801, 0x0001))
 CHESTNUT_ROM_USB_IDS = ((0x174C, 0x2464), (0x174C, 0x2463))
+CHESTNUT_USB_PRODUCT = f"custom {CHESTNUT_FW_VERSION}-CLEAN"
 USB_DEVICES_PATH = Path("/sys/bus/usb/devices")
 TYPEC_CC_ORIENTATION_PATH = Path("/sys/class/power_supply/usb/typec_cc_orientation")
 PRIMARY_USB_CONTROLLER = "a600000.ssusb"
 
 
+def is_chestnut_usb_id(vendor_id: int, product_id: int, include_bootloader: bool = False) -> bool:
+  ids = CHESTNUT_USB_IDS + CHESTNUT_ROM_USB_IDS if include_bootloader else CHESTNUT_USB_IDS
+  return (vendor_id, product_id) in ids
+
+
 def get_usb_topology() -> set[str]:
   try:
     return set(os.listdir(USB_DEVICES_PATH))
@@ -81,7 +87,7 @@ def set_usb_state(device_state, devices: list[dict]) -> None:
     entry.linkErrorCount = device["linkErrorCount"]
     entry.usb3Lane = device.get("usb3Lane", "unknown")
 
-    if (entry.vendorId, entry.productId) in CHESTNUT_USB_IDS:
+    if is_chestnut_usb_id(entry.vendorId, entry.productId):
       chestnut_present = True
 
   device_state.chestnutPresent = chestnut_present

openpilot/common/params_keys.h

@@ -92,6 +92,12 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
     {"ObdMultiplexingEnabled", {CLEAR_ON_MANAGER_START | CLEAR_ON_ONROAD_TRANSITION, BOOL}},
     {"Offroad_CarUnrecognized", {CLEAR_ON_MANAGER_START | CLEAR_ON_ONROAD_TRANSITION, JSON}},
     {"Offroad_ChestnutBranch", {CLEAR_ON_MANAGER_START, JSON}},
+    {"Offroad_ChestnutNotDetected", {CLEAR_ON_MANAGER_START | CLEAR_ON_ONROAD_TRANSITION, JSON}},
+    {"Offroad_ChestnutOverheated", {CLEAR_ON_MANAGER_START, JSON}},
+    {"Offroad_ChestnutPcieUnavailable", {CLEAR_ON_MANAGER_START, JSON}},
+    {"Offroad_ChestnutUncompiled", {CLEAR_ON_MANAGER_START | CLEAR_ON_ONROAD_TRANSITION, JSON}},
+    {"Offroad_ChestnutUpdateFailed", {CLEAR_ON_MANAGER_START | CLEAR_ON_ONROAD_TRANSITION, JSON}},
+    {"Offroad_ChestnutUsbSlow", {CLEAR_ON_MANAGER_START | CLEAR_ON_ONROAD_TRANSITION, JSON}},
     {"Offroad_ConnectivityNeeded", {CLEAR_ON_MANAGER_START, JSON}},
     {"Offroad_ConnectivityNeededPrompt", {CLEAR_ON_MANAGER_START, JSON}},
     {"Offroad_ExcessiveActuation", {PERSISTENT, JSON}},

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_FW_VERSION, CHESTNUT_USB_IDS, USB_DEVICES_PATH
+from openpilot.common.hardware.usb import CHESTNUT_USB_PRODUCT, USB_DEVICES_PATH, is_chestnut_usb_id
 
 MODELS_DIR = Path(__file__).resolve().parent / 'models'
 TG_INPUT_DEVICES_PATH = MODELS_DIR / 'tg_input_devices.json'
@@ -50,7 +50,7 @@ def chestnut_present() -> bool:
     try:
       usb_id = (int((d / "idVendor").read_text(), 16), int((d / "idProduct").read_text(), 16))
       product = (d / "product").read_text().strip()
-      if usb_id in CHESTNUT_USB_IDS and product == f"custom {CHESTNUT_FW_VERSION}-CLEAN":
+      if is_chestnut_usb_id(*usb_id) and product == CHESTNUT_USB_PRODUCT:
         return True
     except Exception:
       pass

openpilot/selfdrive/selfdrived/alerts_offroad.json

@@ -19,6 +19,30 @@
   },
   "Offroad_ChestnutBranch": {
     "text": "Chestnut detected! Switch to the release-chestnut branch to use chestnut-class models.",
+    "severity": -1
+  },
+  "Offroad_ChestnutNotDetected": {
+    "text": "Chestnut not detected. Check USB and 12V connections.",
+    "severity": 0
+  },
+  "Offroad_ChestnutOverheated": {
+    "text": "Chestnut overheated. Ensure good airflow. Current GPU temperature is %1.",
+    "severity": 0
+  },
+  "Offroad_ChestnutPcieUnavailable": {
+    "text": "%1",
+    "severity": 0
+  },
+  "Offroad_ChestnutUncompiled": {
+    "text": "Chestnut model not compiled. Keep ignition on and reboot the comma.",
+    "severity": 0
+  },
+  "Offroad_ChestnutUpdateFailed": {
+    "text": "Chestnut update failed. Check the USB cable.",
+    "severity": 0
+  },
+  "Offroad_ChestnutUsbSlow": {
+    "text": "Chestnut USB link is slow. Check the USB cable. The current speed is %1.",
     "severity": 0
   },
   "Offroad_UnregisteredHardware": {

openpilot/system/hardware/chestnut/status.py

@@ -0,0 +1,96 @@
+import time
+
+from openpilot.common.hardware.usb import CHESTNUT_USB_PRODUCT, is_chestnut_usb_id
+from openpilot.selfdrive.modeld.helpers import chestnut_compiled
+
+
+CHESTNUT_RELEASE_BRANCHES = ("release-chestnut", "release-chestnut-staging")
+CHESTNUT_POWERED_VOLTAGE = 5000
+GPU_TEMP_LIMIT = 100.
+MEMORY_TEMP_LIMIT = 95.
+TEMP_HYSTERESIS = 5.
+
+
+class ChestnutStatus:
+  def __init__(self):
+    self.started = time.monotonic()
+    self.offroad = True
+    self.pcie_failed = False
+    self.power_seen = False
+    self.power_unavailable = False
+    self.power_lost = False
+    self.power_restored = False
+    self.link_failures = 0
+    self.model_loading_seen = False
+    self.model_attempted = False
+    self.overheated = False
+    self.usb_seen = False
+    self.usb_failed = False
+
+  def update(self, offroad: bool, branch: str, usb_state: list[dict], firmware_failed: bool,
+             model_loading: bool, model_active: bool | None, state, set_alert) -> None:
+    detected = [d for d in usb_state if is_chestnut_usb_id(d["vendorId"], d["productId"], include_bootloader=True)]
+    devices = [d for d in detected if is_chestnut_usb_id(d["vendorId"], d["productId"])]
+    firmware_ok = len(devices) == 1 and devices[0]["product"] == CHESTNUT_USB_PRODUCT
+
+    if self.offroad and not offroad:
+      self.pcie_failed = False
+      self.power_seen = False
+      self.power_unavailable = False
+      self.power_lost = False
+      self.power_restored = False
+      self.link_failures = 0
+      self.model_loading_seen = False
+      self.model_attempted = False
+      self.usb_seen = firmware_ok
+      self.usb_failed = False
+
+    self.model_loading_seen |= model_loading
+    self.model_attempted |= self.model_loading_seen and not model_loading and model_active is not None
+
+    if not offroad and self.usb_seen and not firmware_ok:
+      self.usb_failed = True
+
+    if not offroad and state is not None:
+      powered = state.supplyVoltage >= CHESTNUT_POWERED_VOLTAGE
+      power_lost = state.supplyFault or not powered
+      if self.model_attempted and power_lost and not self.power_lost:
+        self.power_unavailable = not self.power_seen
+      self.power_seen |= powered
+
+    if not offroad and self.model_attempted and state is not None:
+      self.link_failures = self.link_failures + 1 if state.pcieLtssm != 0x78 else 0
+      self.pcie_failed |= self.link_failures >= 2 or power_lost
+      self.power_lost |= power_lost
+
+    if self.pcie_failed and self.power_lost and state is not None:
+      self.power_restored |= not state.supplyFault and state.supplyVoltage >= CHESTNUT_POWERED_VOLTAGE
+    if self.usb_failed:
+      self.pcie_failed = False
+      self.power_seen = False
+      self.power_unavailable = False
+      self.power_lost = False
+      self.power_restored = False
+
+    if state is not None:
+      gpu_limit = GPU_TEMP_LIMIT - (TEMP_HYSTERESIS if self.overheated else 0.)
+      memory_limit = MEMORY_TEMP_LIMIT - (TEMP_HYSTERESIS if self.overheated else 0.)
+      self.overheated = state.tempC >= gpu_limit or state.memoryTempC >= memory_limit
+
+    release = branch in CHESTNUT_RELEASE_BRANCHES
+    missing = self.usb_failed or (offroad and release and time.monotonic() - self.started > 10. and len(detected) != 1)
+    slow_usb = offroad and len(devices) == 1 and devices[0]["speedMbps"] < 5000
+    set_alert("Offroad_ChestnutBranch", not release and len(devices) == 1)
+    set_alert("Offroad_ChestnutNotDetected", missing)
+    set_alert("Offroad_ChestnutOverheated", self.overheated, f"{state.tempC:.0f} °C" if state is not None else None)
+    set_alert("Offroad_ChestnutUsbSlow", slow_usb, f"{devices[0]['speedMbps']} Mbps" if slow_usb else None)
+    if self.power_lost:
+      pcie_alert = ("Chestnut power restored. 12V is stable again, cycle ignition." if self.power_restored else
+                    "Chestnut power disconnected. Check 12V connection, then cycle ignition." if self.power_unavailable else
+                    "Chestnut power lost. Possibly caused by an engine-crank voltage drop. Check 12V connection, then cycle ignition.")
+    else:
+      pcie_alert = "Chestnut GPU unavailable. PCIe link is not up. Check the GPU is securely seated."
+    set_alert("Offroad_ChestnutPcieUnavailable", self.pcie_failed, pcie_alert)
+    set_alert("Offroad_ChestnutUncompiled", offroad and firmware_ok and not chestnut_compiled())
+    set_alert("Offroad_ChestnutUpdateFailed", offroad and firmware_failed)
+    self.offroad = offroad

openpilot/system/hardware/hardwared.py

@@ -16,16 +16,17 @@ from openpilot.common.utils import strip_deprecated_keys
 from openpilot.common.filter_simple import FirstOrderFilter
 from openpilot.common.params import Params
 from openpilot.common.realtime import DT_HW
-from openpilot.selfdrive.modeld.helpers import MODELS_DIR, chestnut_compiled
 from openpilot.selfdrive.selfdrived.alertmanager import set_offroad_alert
 from openpilot.common.hardware import HARDWARE, COMMA_HARDWARE, PC
 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.git import get_short_branch
+from openpilot.common.hardware.usb import CHESTNUT_FW_VERSION, CHESTNUT_USB_PRODUCT, get_usb_state, get_usb_topology, is_chestnut_usb_id, set_usb_state
 from openpilot.common.linux import LinuxSystemStats
 from openpilot.system.loggerd.config import get_available_percent
 from openpilot.common.swaglog import cloudlog
 from openpilot.system.hardware.power_monitoring import PowerMonitoring
 from openpilot.system.hardware.fan_controller import FanController
+from openpilot.system.hardware.chestnut.status import ChestnutStatus
 from openpilot.common.version import terms_version, training_version
 from openpilot.system.athena.registration import UNREGISTERED_DONGLE_ID
 
@@ -48,6 +49,11 @@ class Chestnut:
     self.attempts = 0
     self.last_attempt = 0.
     self.flashed = False
+    self.mismatch = False
+
+  @property
+  def failed(self) -> bool:
+    return self.mismatch and self.attempts >= self.MAX_ATTEMPTS and self.thread is not None and not self.thread.is_alive() and not self.flashed
 
   def flash(self) -> None:
     ret = subprocess.run(["sudo", sys.executable, os.path.join(BASEDIR, "openpilot/system/hardware/chestnut/flash.py"), CHESTNUT_FW_VERSION],
@@ -56,9 +62,9 @@ class Chestnut:
     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.mismatch = any(is_chestnut_usb_id(d["vendorId"], d["productId"], include_bootloader=True) and
+                        d["product"] != CHESTNUT_USB_PRODUCT for d in usb_state)
+    if not self.mismatch:
       self.flashed = False
       return
 
@@ -190,7 +196,7 @@ def hw_state_thread(end_event, hw_queue):
 def hardware_thread(end_event, hw_queue) -> None:
   system_stats = LinuxSystemStats()
   pm = messaging.PubMaster(['deviceState'])
-  sm = messaging.SubMaster(["peripheralState", "gpsLocationExternal", "selfdriveState", "pandaStates"], poll="pandaStates")
+  sm = messaging.SubMaster(["peripheralState", "gpsLocationExternal", "selfdriveState", "pandaStates", "chestnutState"], poll="pandaStates")
 
   count = 0
 
@@ -238,7 +244,8 @@ def hardware_thread(end_event, hw_queue) -> None:
 
   fan_controller = FanController(int(1./DT_HW))
   chestnut = Chestnut()
-  big_model_available = (MODELS_DIR / 'big_driving_supercombo.onnx').is_file() or chestnut_compiled()
+  chestnut_status = ChestnutStatus()
+  branch = get_short_branch()
 
   while not end_event.is_set():
     sm.update(PANDA_STATES_TIMEOUT)
@@ -300,8 +307,11 @@ def hardware_thread(end_event, hw_queue) -> None:
 
     set_usb_state(msg.deviceState, last_hw_state.usb_state)
     chestnut.update(started_ts is None, last_hw_state.usb_state)
-    set_offroad_alert_if_changed("Offroad_ChestnutBranch", msg.deviceState.chestnutPresent and not big_model_available)
-
+    chestnut_state = sm["chestnutState"]
+    chestnut_valid = sm.alive["chestnutState"] and sm.valid["chestnutState"]
+    chestnut_status.update(started_ts is None, branch, last_hw_state.usb_state, chestnut.failed,
+                           params.get_bool("ChestnutLoading"), params.get("ChestnutActive"),
+                           chestnut_state if chestnut_valid else None, set_offroad_alert_if_changed)
     # this subset is only used for offroad
     temp_sources = [
       msg.deviceState.memoryTempC,