move to sp

openpilot/selfdrive/ui/sunnypilot/ui_state.py

@@ -66,6 +66,8 @@ class UIStateSP:
     self._sp_initialized: bool = False
 
   def update(self) -> None:
+    self.chestnut_present = self.chestnut_present or self.sm["deviceState"].chestnutPresent
+
     if self.sunnylink_enabled:
       self.sunnylink_state.start()
     else:

openpilot/selfdrive/ui/ui_state.py

@@ -220,7 +220,6 @@ class UIState(UIStateSP):
                              ChestnutState.UNCOMPILED if detected else ChestnutState.DISCONNECTED)
       return
 
-    self.chestnut_present = self.chestnut_present or detected
     model_seen = self.sm.recv_frame["modelV2"] > self.started_frame
     if not self.chestnut_present:
       self.chestnut_state = ChestnutState.DISCONNECTED

openpilot/sunnypilot/system/hardware/aux_power.py

@@ -0,0 +1,37 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+import subprocess
+
+from openpilot.common.params import Params
+from openpilot.common.hardware.usb import CHESTNUT_FW_VERSION, CHESTNUT_ROM_USB_IDS, CHESTNUT_USB_IDS
+from openpilot.system.hardware.chestnut.flash import VBUS_PATH
+
+
+class AuxPower:
+  def __init__(self):
+    self.params = Params()
+    self.vbus_on: bool | None = None
+
+  def _set_vbus(self, on: bool) -> None:
+    if on == self.vbus_on:
+      return
+    subprocess.run(["sudo", "tee", VBUS_PATH], input=b"1" if on else b"0", stdout=subprocess.DEVNULL, check=False)
+    self.vbus_on = on
+
+  def _powersave(self) -> bool:
+    try:
+      return self.params.get_bool("AuxPowerSave")
+    except Exception:
+      return False
+
+  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)
+    self._set_vbus((not offroad or mismatch) or not self._powersave())
+
+
+aux_power = AuxPower()

openpilot/system/hardware/hardwared.py

@@ -21,7 +21,7 @@ from openpilot.selfdrive.selfdrived.alertmanager import set_offroad_alert
 from openpilot.common.hardware import HARDWARE, COMMA_HARDWARE
 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.system.hardware.chestnut.flash import VBUS_PATH
+from openpilot.sunnypilot.system.hardware.aux_power import aux_power
 from openpilot.common.linux import LinuxSystemStats
 from openpilot.system.loggerd.config import get_available_percent
 from openpilot.common.swaglog import cloudlog
@@ -50,10 +50,6 @@ class Chestnut:
     self.attempts = 0
     self.last_attempt = 0.
     self.flashed = False
-    self.vbus_on = None
-    self.params = Params()
-    self.powersave = False
-    self.last_offroad = None
 
   def flash(self) -> None:
     ret = subprocess.run(["sudo", sys.executable, os.path.join(BASEDIR, "openpilot/system/hardware/chestnut/flash.py"), CHESTNUT_FW_VERSION],
@@ -61,19 +57,9 @@ class Chestnut:
     cloudlog.event("chestnut flash done", returncode=ret.returncode, output=ret.stdout[-1000:], error=ret.returncode != 0)
     self.flashed = ret.returncode == 0
 
-  def set_vbus(self, on: bool) -> None:
-    if on == self.vbus_on:
-      return
-    subprocess.run(["sudo", "tee", VBUS_PATH], input=b"1" if on else b"0", stdout=subprocess.DEVNULL, check=False)
-    self.vbus_on = on
-
   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 offroad != self.last_offroad:
-      self.powersave = self.params.get_bool("AuxPowerSave")
-      self.last_offroad = offroad
-    self.set_vbus((not offroad or mismatch) or not self.powersave)
     if not mismatch:
       self.flashed = False
       return
@@ -316,6 +302,7 @@ 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)
+    aux_power.update(started_ts is None, last_hw_state.usb_state)
     current_channel = get_build_metadata().channel
     chestnut_target = CHESTNUT_BRANCHES.get(current_channel)
     chestnut_needs_switch = msg.deviceState.chestnutPresent and not big_model_available and chestnut_target is not None