mochi mochi mochi

selfdrive/car/card.py

@@ -283,9 +283,9 @@ class Car:
     CS.vCruise = float(self.v_cruise_helper.v_cruise_kph)
     CS.vCruiseCluster = float(self.v_cruise_helper.v_cruise_cluster_kph)
 
-    if any(be.type in (ButtonType.accelCruise, ButtonType.resumeCruise) for be in filtered_CS.buttonEvents):
+    if any(be.type in (ButtonType.accelCruise, ButtonType.accelHardCruise, ButtonType.resumeCruise) for be in filtered_CS.buttonEvents):
       self.resume_prev_button = True
-    elif any(be.type in (ButtonType.decelCruise, ButtonType.setCruise) for be in filtered_CS.buttonEvents):
+    elif any(be.type in (ButtonType.decelCruise, ButtonType.decelHardCruise, ButtonType.setCruise) for be in filtered_CS.buttonEvents):
       self.resume_prev_button = False
 
     FPCS = self.starpilot_card.update(CS, FPCS, self.sm, self.starpilot_toggles)

selfdrive/car/cruise.py

@@ -24,11 +24,17 @@ CRUISE_LONG_PRESS = 50
 CRUISE_NEAREST_FUNC = {
   ButtonType.accelCruise: math.ceil,
   ButtonType.decelCruise: math.floor,
+  ButtonType.accelHardCruise: math.ceil,
+  ButtonType.decelHardCruise: math.floor,
 }
 CRUISE_INTERVAL_SIGN = {
   ButtonType.accelCruise: +1,
   ButtonType.decelCruise: -1,
+  ButtonType.accelHardCruise: +1,
+  ButtonType.decelHardCruise: -1,
 }
+HARD_CRUISE_BUTTONS = (ButtonType.accelHardCruise, ButtonType.decelHardCruise)
+ACCEL_CRUISE_BUTTONS = (ButtonType.accelCruise, ButtonType.accelHardCruise)
 
 
 class VCruiseHelper:
@@ -37,7 +43,12 @@ class VCruiseHelper:
     self.v_cruise_kph = V_CRUISE_UNSET
     self.v_cruise_cluster_kph = V_CRUISE_UNSET
     self.v_cruise_kph_last = 0
-    self.button_timers = {ButtonType.decelCruise: 0, ButtonType.accelCruise: 0}
+    self.button_timers = {
+      ButtonType.decelCruise: 0,
+      ButtonType.accelCruise: 0,
+      ButtonType.decelHardCruise: 0,
+      ButtonType.accelHardCruise: 0,
+    }
     self.button_change_states = {btn: {"standstill": False, "enabled": False} for btn in self.button_timers}
 
     self.gm_cc_only = self.CP.carFingerprint in CC_ONLY_CAR and self.CP.flags & GMFlags.CC_LONG.value
@@ -121,7 +132,7 @@ class VCruiseHelper:
 
     # Don't adjust speed when pressing resume to exit standstill
     cruise_standstill = self.button_change_states[button_type]["standstill"] or CS.cruiseState.standstill
-    if button_type == ButtonType.accelCruise and cruise_standstill:
+    if button_type in ACCEL_CRUISE_BUTTONS and cruise_standstill:
       return
 
     # Don't adjust speed if we've enabled since the button was depressed (some ports enable on rising edge)
@@ -129,7 +140,7 @@ class VCruiseHelper:
       return
 
     short_interval, long_interval = self._get_cruise_delta_intervals(starpilot_toggles)
-    v_cruise_delta_interval = long_interval if long_press else short_interval
+    v_cruise_delta_interval = long_interval if long_press or button_type in HARD_CRUISE_BUTTONS else short_interval
     v_cruise_delta = v_cruise_delta * v_cruise_delta_interval
     if v_cruise_delta_interval % 5 == 0 and self.v_cruise_kph % v_cruise_delta != 0:  # partial interval
       self.v_cruise_kph = CRUISE_NEAREST_FUNC[button_type](self.v_cruise_kph / v_cruise_delta) * v_cruise_delta
@@ -137,7 +148,7 @@ class VCruiseHelper:
       self.v_cruise_kph += v_cruise_delta * CRUISE_INTERVAL_SIGN[button_type]
 
     # If set is pressed while overriding, clip cruise speed to minimum of vEgo
-    if CS.gasPressed and button_type in (ButtonType.decelCruise, ButtonType.setCruise):
+    if CS.gasPressed and button_type in (ButtonType.decelCruise, ButtonType.decelHardCruise, ButtonType.setCruise):
       self.v_cruise_kph = max(self.v_cruise_kph, CS.vEgo * CV.MS_TO_KPH)
 
     self.v_cruise_kph = np.clip(round(self.v_cruise_kph, 1), V_CRUISE_MIN, V_CRUISE_MAX)
@@ -161,7 +172,7 @@ class VCruiseHelper:
       return
 
     engage_floor_kph = max(V_CRUISE_MIN, 7.0 * CV.MPH_TO_KPH)
-    resume_pressed = any(b.type in (ButtonType.accelCruise, ButtonType.resumeCruise) for b in CS.buttonEvents)
+    resume_pressed = any(b.type in (ButtonType.accelCruise, ButtonType.accelHardCruise, ButtonType.resumeCruise) for b in CS.buttonEvents)
     remembered_resume = resume_prev_button and (self.gm_cc_only or self.redneck_non_pcm)
 
     if self.v_cruise_initialized and (resume_pressed or remembered_resume):

selfdrive/car/tests/test_cruise_speed.py

@@ -1,5 +1,6 @@
 import pytest
 import itertools
+import math
 import numpy as np
 
 from parameterized import parameterized_class
@@ -101,6 +102,62 @@ class TestVCruiseHelper:
         )
         assert pressed == (self.v_cruise_helper.v_cruise_kph == self.v_cruise_helper.v_cruise_kph_last)
 
+  def test_hard_press_uses_long_press_interval(self):
+    self.enable(52 * CV.MPH_TO_MS, False)
+    initial_v_cruise_kph = self.v_cruise_helper.v_cruise_kph
+
+    pressed_cs = car.CarState(cruiseState={"available": True})
+    pressed_cs.buttonEvents = [ButtonEvent(type=ButtonType.accelHardCruise, pressed=True)]
+    self.v_cruise_helper.update_v_cruise(
+      pressed_cs,
+      enabled=True,
+      is_metric=False,
+      speed_limit_changed=False,
+      starpilot_toggles=self.starpilot_toggles,
+    )
+
+    released_cs = car.CarState(cruiseState={"available": True})
+    released_cs.buttonEvents = [ButtonEvent(type=ButtonType.accelHardCruise, pressed=False)]
+    self.v_cruise_helper.update_v_cruise(
+      released_cs,
+      enabled=True,
+      is_metric=False,
+      speed_limit_changed=False,
+      starpilot_toggles=self.starpilot_toggles,
+    )
+
+    hard_interval = self.starpilot_toggles.cruise_increase_long * IMPERIAL_INCREMENT
+    expected_kph = math.ceil(initial_v_cruise_kph / hard_interval) * hard_interval
+    assert self.v_cruise_helper.v_cruise_kph == pytest.approx(expected_kph)
+
+  def test_hard_decel_press_uses_long_press_interval(self):
+    self.enable(52 * CV.MPH_TO_MS, False)
+    initial_v_cruise_kph = self.v_cruise_helper.v_cruise_kph
+
+    pressed_cs = car.CarState(cruiseState={"available": True})
+    pressed_cs.buttonEvents = [ButtonEvent(type=ButtonType.decelHardCruise, pressed=True)]
+    self.v_cruise_helper.update_v_cruise(
+      pressed_cs,
+      enabled=True,
+      is_metric=False,
+      speed_limit_changed=False,
+      starpilot_toggles=self.starpilot_toggles,
+    )
+
+    released_cs = car.CarState(cruiseState={"available": True})
+    released_cs.buttonEvents = [ButtonEvent(type=ButtonType.decelHardCruise, pressed=False)]
+    self.v_cruise_helper.update_v_cruise(
+      released_cs,
+      enabled=True,
+      is_metric=False,
+      speed_limit_changed=False,
+      starpilot_toggles=self.starpilot_toggles,
+    )
+
+    hard_interval = self.starpilot_toggles.cruise_increase_long * IMPERIAL_INCREMENT
+    expected_kph = math.floor(initial_v_cruise_kph / hard_interval) * hard_interval
+    assert self.v_cruise_helper.v_cruise_kph == pytest.approx(expected_kph)
+
   def test_rising_edge_enable(self):
     """
     Some car interfaces may enable on rising edge of a button,

selfdrive/selfdrived/helpers.py

@@ -22,11 +22,15 @@ class ExcessiveActuationCheck:
     self._excessive_counter = 0
     self._engaged_counter = 0
 
-  def update(self, sm: messaging.SubMaster, CS: car.CarState, calibrated_pose: Pose) -> ExcessiveActuationType | None:
+  def update(self, sm: messaging.SubMaster, CS: car.CarState, calibrated_pose: Pose,
+             allow_impossible_acceleration: bool = False) -> ExcessiveActuationType | None:
     # CS.aEgo can be noisy to bumps in the road, transitioning from standstill, losing traction, etc.
     # longitudinal
     accel_calibrated = calibrated_pose.acceleration.x
-    excessive_long_actuation = sm['carControl'].longActive and (accel_calibrated > ACCEL_MAX * 2 or accel_calibrated < ACCEL_MIN * 2)
+    excessive_long_actuation = (
+      not allow_impossible_acceleration and sm['carControl'].longActive and
+      (accel_calibrated > ACCEL_MAX * 2 or accel_calibrated < ACCEL_MIN * 2)
+    )
 
     # lateral
     yaw_rate = calibrated_pose.angular_velocity.yaw

selfdrive/selfdrived/selfdrived.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+import json
 import os
 import time
 import threading
@@ -71,6 +72,9 @@ class SelfdriveD:
     self.pose_calibrator = PoseCalibrator()
     self.calibrated_pose: Pose | None = None
     self.excessive_actuation_check = ExcessiveActuationCheck()
+    self.allow_impossible_acceleration = self.params.get_bool("AllowImpossibleAcceleration")
+    if self.allow_impossible_acceleration:
+      self.clear_longitudinal_excessive_actuation_alert()
     self.excessive_actuation = self.params.get("Offroad_ExcessiveActuation") is not None
 
     # Setup sockets
@@ -172,6 +176,32 @@ class SelfdriveD:
 
     self.FPCP = messaging.log_from_bytes(self.params.get("StarPilotCarParams", block=True), custom.StarPilotCarParams)
 
+  def clear_longitudinal_excessive_actuation_alert(self):
+    alert = self.params.get("Offroad_ExcessiveActuation")
+    if not alert:
+      return
+
+    if isinstance(alert, bytes):
+      try:
+        alert = json.loads(alert.decode("utf-8", errors="replace"))
+      except json.JSONDecodeError:
+        return
+    elif isinstance(alert, str):
+      try:
+        alert = json.loads(alert)
+      except json.JSONDecodeError:
+        return
+
+    if not isinstance(alert, dict):
+      return
+
+    extra = alert.get("extra", "")
+    if isinstance(extra, bytes):
+      extra = extra.decode("utf-8", errors="replace")
+
+    if str(extra).strip().lower() == "longitudinal":
+      self.params.remove("Offroad_ExcessiveActuation")
+
   def update_events(self, CS):
     """Compute onroadEvents from carState"""
 
@@ -221,7 +251,7 @@ class SelfdriveD:
       return
 
     # Block resume if cruise never previously enabled
-    resume_pressed = any(be.type in (ButtonType.accelCruise, ButtonType.resumeCruise) for be in CS.buttonEvents)
+    resume_pressed = any(be.type in (ButtonType.accelCruise, ButtonType.accelHardCruise, ButtonType.resumeCruise) for be in CS.buttonEvents)
     if not self.CP.pcmCruise and CS.vCruise > 250 and resume_pressed:
       self.events.add(EventName.resumeBlocked)
 
@@ -332,7 +362,9 @@ class SelfdriveD:
       self.calibrated_pose = self.pose_calibrator.build_calibrated_pose(device_pose)
 
     if self.calibrated_pose is not None:
-      excessive_actuation = self.excessive_actuation_check.update(self.sm, CS, self.calibrated_pose)
+      excessive_actuation = self.excessive_actuation_check.update(
+        self.sm, CS, self.calibrated_pose, self.allow_impossible_acceleration
+      )
       if not self.excessive_actuation and excessive_actuation is not None:
         set_offroad_alert("Offroad_ExcessiveActuation", True, extra_text=str(excessive_actuation))
         self.excessive_actuation = True