From 42057a3f8f3bba685a2aab0dcd22b3ad64e0b7db Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Thu, 29 Oct 2026 14:33:29 -0500 Subject: [PATCH] Limit low-speed steering advisories to engaged or armed steering --- openpilot/selfdrive/selfdrived/selfdrived.py | 12 ++- .../starpilot/lateral/low_speed_advisory.py | 10 ++- .../lateral/tests/test_low_speed_advisory.py | 89 +++++++++++++++++++ prebuilt.json | 2 +- 4 files changed, 109 insertions(+), 4 deletions(-) diff --git a/openpilot/selfdrive/selfdrived/selfdrived.py b/openpilot/selfdrive/selfdrived/selfdrived.py index 47d8d28cad..e569149579 100644 --- a/openpilot/selfdrive/selfdrived/selfdrived.py +++ b/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) diff --git a/openpilot/starpilot/lateral/low_speed_advisory.py b/openpilot/starpilot/lateral/low_speed_advisory.py index c2b9963fc5..2620575f05 100644 --- a/openpilot/starpilot/lateral/low_speed_advisory.py +++ b/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 diff --git a/openpilot/starpilot/lateral/tests/test_low_speed_advisory.py b/openpilot/starpilot/lateral/tests/test_low_speed_advisory.py index fb4469112c..55bd14451d 100644 --- a/openpilot/starpilot/lateral/tests/test_low_speed_advisory.py +++ b/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') diff --git a/prebuilt.json b/prebuilt.json index c75542fd01..11efe703a8 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "a8effb6948a39573af7fc459a058f8cc4035777e04053e64e4404566be52d76a", "schema_version": 1} +{"protected_inventory_sha256": "845201e8021f9955dae8eb23a1c053d4a13e609c70b9d29fac4156450fe8f365", "schema_version": 1}