fix socket

selfdrive/modeld/helpers.py

@@ -8,6 +8,7 @@ import tempfile
 from pathlib import Path
 
 from tinygrad.device import Device
+from openpilot.system.hardware.usb import chestnut_present
 
 MODELS_DIR = Path(__file__).resolve().parent / "models"
 TG_INPUT_DEVICES_PATH = MODELS_DIR / "tg_input_devices.json"
@@ -69,14 +70,7 @@ def modeld_pkl_path(usbgpu: bool) -> Path:
 
 
 def usbgpu_present() -> bool:
-  for d in Path("/sys/bus/usb/devices").glob("*"):
-    try:
-      if int((d / "idVendor").read_text(), 16) == USBGPU_VID and \
-          int((d / "idProduct").read_text(), 16) == USBGPU_PID:
-        return True
-    except Exception:
-      pass
-  return False
+  return chestnut_present()
 
 
 def tinygrad_dev_config(usbgpu: bool, tici: bool) -> str:

selfdrive/ui/ui_state.py

@@ -1,7 +1,6 @@
 import json
 import pyray as rl
 import numpy as np
-import socket
 import time
 import threading
 from collections.abc import Callable
@@ -11,13 +10,12 @@ from openpilot.common.filter_simple import FirstOrderFilter
 from openpilot.common.params import Params
 from openpilot.common.swaglog import cloudlog
 from openpilot.selfdrive.ui.lib.prime_state import PrimeState
-from openpilot.selfdrive.modeld.helpers import usbgpu_present
+from openpilot.system.hardware.usb import chestnut_present
 from openpilot.system.ui.lib.application import gui_app
 from openpilot.system.hardware import HARDWARE, PC
 
-NETLINK_KOBJECT_UEVENT = 15
-
 BACKLIGHT_OFFROAD = 65 if HARDWARE.get_device_type() == "mici" else 50
+USBGPU_POLL_INTERVAL = 1.0
 
 
 class UIStatus(Enum):
@@ -85,20 +83,10 @@ class UIState:
     self.is_metric: bool = self.params.get_bool("IsMetric")
     self.is_release = self.params.get_bool("IsReleaseBranch")
     self.always_on_dm: bool = self.params.get_bool("AlwaysOnDM")
-    self.usbgpu: bool = usbgpu_present()
+    self.usbgpu: bool = chestnut_present()
     self.usbgpu_compiled: bool = self.params.get_bool("UsbGpuCompiled")
     self.usbgpu_active: bool = self.params.get_bool("UsbGpuActive")
-    self._uevents: socket.socket | None = None
-    if hasattr(socket, "AF_NETLINK"):
-      try:
-        self._uevents = socket.socket(socket.AF_NETLINK, socket.SOCK_DGRAM, NETLINK_KOBJECT_UEVENT)
-        self._uevents.bind((0, 1))
-        self._uevents.setblocking(False)
-      except OSError:
-        cloudlog.exception("failed to monitor USB hotplug events")
-        if self._uevents is not None:
-          self._uevents.close()
-          self._uevents = None
+    self._usbgpu_update_time: float = 0.0
     self.started: bool = False
     self.ignition: bool = False
     self.recording_audio: bool = False
@@ -200,16 +188,10 @@ class UIState:
 
     self.is_metric = self.params.get_bool("IsMetric")
     self.always_on_dm = self.params.get_bool("AlwaysOnDM")
-    if self._uevents is not None:
-      try:
-        while self._uevents.recv(4096):
-          self.usbgpu = usbgpu_present()
-      except BlockingIOError:
-        pass
-      except OSError:
-        cloudlog.exception("failed to read USB hotplug event")
-        self._uevents.close()
-        self._uevents = None
+    now = time.monotonic()
+    if now - self._usbgpu_update_time >= USBGPU_POLL_INTERVAL:
+      self.usbgpu = chestnut_present()
+      self._usbgpu_update_time = now
     self.usbgpu_compiled = self.params.get_bool("UsbGpuCompiled")
     self.usbgpu_active = self.params.get_bool("UsbGpuActive")
     self.switchback_mode_enabled = self.params_memory.get_bool("SwitchbackModeEnabled") if self.started else False

system/hardware/tests/test_usb.py

@@ -0,0 +1,21 @@
+from openpilot.system.hardware import usb
+
+
+def test_chestnut_present(tmp_path, monkeypatch):
+  monkeypatch.setattr(usb, "USB_DEVICES_PATH", tmp_path)
+  device = tmp_path / "1-1"
+  device.mkdir()
+  (device / "idVendor").write_text("add1\n")
+  (device / "idProduct").write_text("0001\n")
+
+  assert usb.chestnut_present()
+
+
+def test_chestnut_absent_for_other_usb_device(tmp_path, monkeypatch):
+  monkeypatch.setattr(usb, "USB_DEVICES_PATH", tmp_path)
+  device = tmp_path / "1-1"
+  device.mkdir()
+  (device / "idVendor").write_text("18d1\n")
+  (device / "idProduct").write_text("4ee7\n")
+
+  assert not usb.chestnut_present()

system/hardware/usb.py

@@ -20,6 +20,14 @@ def usb_devices() -> list[Path]:
     return []
 
 
+def chestnut_present() -> bool:
+  return any(
+    read_int(device / "idVendor", 16) == CHESTNUT_VENDOR_ID and
+    read_int(device / "idProduct", 16) == CHESTNUT_PRODUCT_ID
+    for device in usb_devices()
+  )
+
+
 def controller(device: Path) -> Path | None:
   try:
     return next((parent for parent in device.resolve().parents if parent.name.endswith(".ssusb")), None)