Limit low-speed steering advisories to engaged or armed steering

openpilot/selfdrive/selfdrived/selfdrived.py

@@ -58,7 +58,7 @@ from openpilot.starpilot.conditional_mode.runtime_settings import (ConditionalSe
 from openpilot.starpilot.conditional_mode.status import settings_fingerprint
 from openpilot.starpilot.feature_runtime import enabled as feature_enabled
 from openpilot.starpilot.nostalgia import aol_no_entry, paddle_cancel, physical_cancel, saved_enabled as nostalgia_saved_enabled
-from openpilot.starpilot.lateral.low_speed_advisory import LowSpeedAdvisory
+from openpilot.starpilot.lateral.low_speed_advisory import LowSpeedAdvisory, advisory_eligible
 from openpilot.starpilot.lateral.lane_change_status_wire import alert_wording, decode as decode_lane_status, fresh_for_model
 from openpilot.selfdrive.selfdrived.alertmanager import AlertManager, set_offroad_alert
 
@@ -310,7 +310,10 @@ class SelfdriveD:
       device = self.sm['deviceState']
       drive_id = (int(device.startedMonoTime) if self.sm.seen['deviceState'] and self.sm.alive['deviceState'] and
                   self.sm.valid['deviceState'] and device.started else 0)
-      show_low_speed = self.low_speed_advisory.update(float(CS.vEgo), float(self.CP.minSteerSpeed), drive_id=drive_id)
+      eligible = advisory_eligible(self.CP, CS, active=self.active,
+        intent=self.aol_last_intent if self.aol_replay else None)
+      show_low_speed = self.low_speed_advisory.update(float(CS.vEgo), float(self.CP.minSteerSpeed),
+        drive_id=drive_id, eligible=eligible)
       self.events.add_from_msg([event for event in car_events if event.name != EventName.belowSteerSpeed])
       if show_low_speed:
         self.events.add(EventName.belowSteerSpeed)
@@ -1076,6 +1079,11 @@ class SelfdriveD:
     return intent
 
   def update_alerts(self, CS):
+    # Engagement may change after car events; retire only the inactive advisory.
+    if (EventName.belowSteerSpeed in self.events.names or 'belowSteerSpeed/warning' in self.AM.alerts) and not advisory_eligible(
+        self.CP, CS, active=self.active, intent=self.aol_last_intent if self.aol_replay else None):
+      self.events.events = [event for event in self.events.names if event != EventName.belowSteerSpeed]
+      self.AM.alerts.pop('belowSteerSpeed/warning', None)
     clear_event_types = set()
     if ET.WARNING not in self.state_machine.current_alert_types:
       clear_event_types.add(ET.WARNING)

openpilot/starpilot/lateral/low_speed_advisory.py

@@ -3,6 +3,11 @@
 import math
 
 
+def advisory_eligible(CP, CS, *, active: bool, intent) -> bool:
+  armed = intent is not None and intent.allowedLatch and intent.lateralArmed and not intent.pauseLateral
+  return bool((active or armed) and (CP.brand != 'gm' or CS.cruiseState.available))
+
+
 class LowSpeedAdvisory:
   def __init__(self):
     self.drive_id = 0
@@ -10,13 +15,16 @@ class LowSpeedAdvisory:
     self.shown = False
     self.showing = False
 
-  def update(self, speed: float, minimum: float, *, drive_id: int = 0) -> bool:
+  def update(self, speed: float, minimum: float, *, drive_id: int = 0, eligible: bool = True) -> bool:
     # Unknown/transient device state does not restart the advisory lifecycle.
     if drive_id > 0 and drive_id != self.drive_id:
       self.drive_id = drive_id
       self.has_been_above = False
       self.shown = False
       self.showing = False
+    if not eligible:
+      self.showing = False
+      return False
     if not math.isfinite(speed) or not math.isfinite(minimum) or minimum <= 0:
       return False
     under = speed < minimum

openpilot/starpilot/lateral/tests/test_low_speed_advisory.py

@@ -56,6 +56,7 @@ class TestLowSpeedAdvisory(unittest.TestCase):
     with OpenpilotPrefix(), patch('openpilot.selfdrive.selfdrived.selfdrived.REPLAY', True):
       drive = SelfdriveD(CP=cp)
       drive.initialized = True
+      drive.active = True
       device = messaging.new_message('deviceState', valid=True, logMonoTime=1_000_000_000)
       device.deviceState.started = True
       device.deviceState.startedMonoTime = 1_000_000_000
@@ -98,3 +99,91 @@ class TestLowSpeedAdvisory(unittest.TestCase):
       self.assertNotIn(EventName.belowSteerSpeed, sample(1.))
       self.assertNotIn(EventName.belowSteerSpeed, sample(cp.minSteerSpeed))
       self.assertIn(EventName.belowSteerSpeed, sample(1.))
+
+  def test_disarmed_crossings_do_not_spend_the_advisory(self):
+    owner = LowSpeedAdvisory()
+    for speed in (4., 2., 4., 0.):
+      self.assertFalse(owner.update(speed, 3., drive_id=10, eligible=False))
+    self.assertFalse(owner.update(4., 3., drive_id=10, eligible=True))
+    self.assertTrue(owner.update(2., 3., drive_id=10, eligible=True))
+    self.assertFalse(owner.update(2., 3., drive_id=10, eligible=False))
+    self.assertFalse(owner.update(2., 3., drive_id=10, eligible=True))
+
+  def test_advisory_eligibility_keeps_gm_main_and_independent_aol_distinct(self):
+    from types import SimpleNamespace
+    from openpilot.starpilot.lateral.low_speed_advisory import advisory_eligible
+    state = SimpleNamespace(cruiseState=SimpleNamespace(available=False))
+    armed = SimpleNamespace(allowedLatch=True, lateralArmed=True, pauseLateral=False)
+    for brand in ('gm', 'hyundai', 'toyota', 'ford', 'honda'):
+      cp = SimpleNamespace(brand=brand)
+      self.assertFalse(advisory_eligible(cp, state, active=False, intent=None))
+      self.assertEqual(advisory_eligible(cp, state, active=True, intent=None), brand != 'gm')
+      self.assertEqual(advisory_eligible(cp, state, active=False, intent=armed), brand != 'gm')
+    state.cruiseState.available = True
+    for active, latched, lateral, paused, expected in (
+        (False, False, False, False, False), (False, False, True, False, False),
+        (False, True, False, False, False), (False, True, True, True, False),
+        (False, True, True, False, True), (True, False, False, False, True)):
+      intent = SimpleNamespace(allowedLatch=latched, lateralArmed=lateral, pauseLateral=paused)
+      self.assertEqual(advisory_eligible(SimpleNamespace(brand='gm'), state, active=active, intent=intent), expected)
+
+  def test_actual_sierra_events_and_alerts_stay_quiet_with_main_off(self):
+    from types import SimpleNamespace
+    from unittest.mock import patch
+    from opendbc.car import gen_empty_fingerprint, structs
+    from opendbc.car.gm.interface import CarInterface
+    from opendbc.car.gm.values import CAR
+    from openpilot.cereal import messaging
+    from openpilot.common.prefix import OpenpilotPrefix
+    from openpilot.selfdrive.selfdrived.events import ET, EventName
+    from openpilot.selfdrive.selfdrived.selfdrived import SelfdriveD
+
+    fingerprint = gen_empty_fingerprint()
+    fingerprint[0][0xBE] = 6
+    fingerprint[2].update({0x180: 4, 0x320: 6, 0x370: 6})
+    cp = CarInterface.get_params(CAR.CHEVROLET_SILVERADO, fingerprint, [], False, False, False)
+    self.assertAlmostEqual(cp.minSteerSpeed, 10 / 3.6, delta=1e-6)
+    with OpenpilotPrefix(), patch('openpilot.selfdrive.selfdrived.selfdrived.REPLAY', True):
+      drive = SelfdriveD(CP=cp)
+      drive.initialized = True
+      drive.active = drive.enabled = False
+      drive.aol_replay = True
+      device = messaging.new_message('deviceState', valid=True, logMonoTime=1_000_000_000)
+      device.deviceState.started = True
+      device.deviceState.startedMonoTime = 1_000_000_000
+      drive.sm.update_msgs(1., [device.as_reader()])
+      state = structs.CarState.new_message(canValid=True, gearShifter='drive', vCruise=40.)
+      state.cruiseState.available = False
+      previous = state.as_reader()
+
+      def sample(speed):
+        state.vEgo = speed
+        state.lowSpeedAlert = speed < cp.minSteerSpeed
+        drive.sm.frame += 1
+        drive.events.clear()
+        drive._update_car_event_sample(state, previous)
+        # An unrelated AOL pause may request warning presentation while off.
+        drive.state_machine.current_alert_types = [ET.PERMANENT, ET.WARNING, ET.IMMEDIATE_DISABLE]
+        drive.update_alerts(state)
+
+      for speed in (4., 1., 0., 4., 1.):
+        sample(speed)
+        self.assertNotIn(EventName.belowSteerSpeed, drive.events.names)
+        self.assertNotIn('belowSteerSpeed/warning', drive.AM.alerts)
+      state.cruiseState.available = True
+      drive.aol_last_intent = SimpleNamespace(allowedLatch=True, lateralArmed=True, pauseLateral=False)
+      sample(4.)
+      sample(1.)
+      self.assertIn(EventName.belowSteerSpeed, drive.events.names)
+      self.assertEqual(drive.AM.current_alert.alert_type, 'belowSteerSpeed/warning')
+      # Main can turn off after the event sample but before alert selection.
+      state.cruiseState.available = False
+      drive.update_alerts(state)
+      self.assertNotIn(EventName.belowSteerSpeed, drive.events.names)
+      self.assertNotIn('belowSteerSpeed/warning', drive.AM.alerts)
+      for fault in (EventName.steerTempUnavailableSilent, EventName.steerUnavailable, EventName.canError):
+        drive.events.clear()
+        drive.events.add(fault)
+        drive.update_alerts(state)
+        self.assertIn(fault, drive.events.names)
+        self.assertNotEqual(drive.AM.current_alert.alert_type, 'belowSteerSpeed/warning')

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "a8effb6948a39573af7fc459a058f8cc4035777e04053e64e4404566be52d76a", "schema_version": 1}
+{"protected_inventory_sha256": "845201e8021f9955dae8eb23a1c053d4a13e609c70b9d29fac4156450fe8f365", "schema_version": 1}