From fae3f5ccfc20b6ebffd7586b42286f7e355c3033 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Fri, 30 Oct 2026 08:37:20 -0500 Subject: [PATCH] Blindspot alert --- openpilot/common/params_keys.h | 2 + openpilot/selfdrive/selfdrived/selfdrived.py | 14 +++ openpilot/starpilot/audio/blindspot_alert.py | 32 +++++ .../audio/tests/test_blindspot_alert.py | 119 ++++++++++++++++++ openpilot/starpilot/ui/sounds_owner.py | 18 +++ .../ui/tests/test_sounds_settings.py | 4 +- 6 files changed, 187 insertions(+), 2 deletions(-) create mode 100644 openpilot/starpilot/audio/blindspot_alert.py create mode 100644 openpilot/starpilot/audio/tests/test_blindspot_alert.py diff --git a/openpilot/common/params_keys.h b/openpilot/common/params_keys.h index 0dcbd759de..ff22255b80 100644 --- a/openpilot/common/params_keys.h +++ b/openpilot/common/params_keys.h @@ -135,6 +135,8 @@ inline static std::unordered_map keys = { {"LeadInfo", {PERSISTENT, BOOL, "0"}}, {"LeadInfoMode", {PERSISTENT, INT, "2"}}, {"BlindSpotMetrics", {PERSISTENT, BOOL, "1"}}, + {"LoudBlindspotAlert", {PERSISTENT, BOOL, "0"}}, + {"LoudBlindspotAlertWhenDisengaged", {PERSISTENT, BOOL, "0"}}, {"SignalMetrics", {PERSISTENT, BOOL, "0"}}, {"PIPPreviewEnabled", {PERSISTENT, BOOL, "0"}}, {"PIPPreviewMask", {PERSISTENT | DONT_LOG, JSON, "{\"width\":1928,\"height\":1208,\"center_left\":[315,548],\"center_right\":[1571,539],\"crop_size\":580}"}}, diff --git a/openpilot/selfdrive/selfdrived/selfdrived.py b/openpilot/selfdrive/selfdrived/selfdrived.py index 59538c7fe2..b04a7e1281 100755 --- a/openpilot/selfdrive/selfdrived/selfdrived.py +++ b/openpilot/selfdrive/selfdrived/selfdrived.py @@ -207,6 +207,8 @@ class SelfdriveD: self.is_ldw_enabled = self.params.get_bool("IsLdwEnabled") self.disengage_on_accelerator = self.params.get_bool("DisengageOnAccelerator") self.radar_recovery_enabled = self.params.get_bool("RadarRecoveryAlert") + self.blindspot_alert_settings = (self.params.get_bool("LoudBlindspotAlert"), + self.params.get_bool("LoudBlindspotAlertWhenDisengaged")) car_recognized = self.CP.brand != 'mock' @@ -1108,6 +1110,16 @@ class SelfdriveD: if alert.event_type in self.state_machine.current_alert_types) now_ns = time.monotonic_ns() drive = int(self.sm['deviceState'].startedMonoTime) if self.sm['deviceState'].started else 0 + from openpilot.starpilot.audio.blindspot_alert import ALERT_TYPE as BSM_ALERT, blindspot_alert + loud, without_lateral = getattr(self, 'blindspot_alert_settings', (False, False)) + blindspot = blindspot_alert(CS, self.sm, loud=loud, without_lateral=without_lateral, + lane_change_blocked=EventName.laneChangeBlocked in self.events.names, now_ns=now_ns) + if blindspot is not None: + alerts = [alert for alert in alerts if alert.alert_type != 'laneChangeBlocked/warning'] + self.AM.alerts.pop('laneChangeBlocked/warning', None) + alerts.append(blindspot) + else: + self.AM.alerts.pop(BSM_ALERT, None) switchback = bool(self.switchback_capable and self.sm.seen['slcState'] and self.sm.alive['slcState'] and self.sm.valid['slcState'] and self.sm['carControl'].latActive and self.switchback_status.sample_state(self.sm['slcState'], drive_id=drive, now_ns=now_ns, @@ -2168,6 +2180,8 @@ class SelfdriveD: self.is_ldw_enabled = self.params.get_bool("IsLdwEnabled") self.disengage_on_accelerator = self.params.get_bool("DisengageOnAccelerator") self.radar_recovery_enabled = self.params.get_bool("RadarRecoveryAlert") + self.blindspot_alert_settings = (self.params.get_bool("LoudBlindspotAlert"), + self.params.get_bool("LoudBlindspotAlertWhenDisengaged")) self.nostalgia_enabled = nostalgia_saved_enabled(self.params) self.refresh_saved_driving_mode(time.monotonic_ns()) self.personality = self.params.get("LongitudinalPersonality", return_default=True) diff --git a/openpilot/starpilot/audio/blindspot_alert.py b/openpilot/starpilot/audio/blindspot_alert.py new file mode 100644 index 0000000000..362317011d --- /dev/null +++ b/openpilot/starpilot/audio/blindspot_alert.py @@ -0,0 +1,32 @@ +"""Optional blind-spot presentation; never changes lane-change or control authority.""" + +KEYS = ("LoudBlindspotAlert", "LoudBlindspotAlertWhenDisengaged") +ALERT_TYPE = "blindSpotAdvisory/warning" +EVENT_TYPE = "blindSpotAdvisory" + + +def blindspot_alert(cs, sm, *, loud: bool, without_lateral: bool, lane_change_blocked: bool, now_ns: int): + if not loud and not without_lateral: + return None + try: + if not sm['deviceState'].started or not sm.all_checks(['deviceState', 'carState', 'carControl']): + return None + if any(not 0 < int(sm.logMonoTime[key]) <= now_ns or now_ns - int(sm.logMonoTime[key]) > limit + for key, limit in (('deviceState', 2_000_000_000), ('carState', 200_000_000), ('carControl', 200_000_000))): + return None + if not cs.canValid or cs.canTimeout: + return None + matching_signal = (bool(cs.leftBlinker) != bool(cs.rightBlinker) and + (cs.leftBlinker and cs.leftBlindspot or cs.rightBlinker and cs.rightBlindspot)) + if not (loud and lane_change_blocked or + without_lateral and not sm['carControl'].latActive and matching_signal): + return None + except (AttributeError, KeyError, TypeError, ValueError, OverflowError): + return None + from openpilot.selfdrive.selfdrived.events import Alert, AlertSize, AlertStatus, AudibleAlert, Priority, VisualAlert + alert = Alert("Car Detected in Blindspot", "", AlertStatus.userPrompt, AlertSize.small, + Priority.LOW, VisualAlert.none, AudibleAlert.warningSoft, .1) + # Independent of engagement's WARNING filter so the parked/offroad checks, + # not lateral activation, determine eligibility for this optional advisory. + alert.alert_type, alert.event_type = ALERT_TYPE, EVENT_TYPE + return alert diff --git a/openpilot/starpilot/audio/tests/test_blindspot_alert.py b/openpilot/starpilot/audio/tests/test_blindspot_alert.py new file mode 100644 index 0000000000..8cbfe1a5c2 --- /dev/null +++ b/openpilot/starpilot/audio/tests/test_blindspot_alert.py @@ -0,0 +1,119 @@ +from types import SimpleNamespace as NS + +import pytest + +from openpilot.starpilot.audio.blindspot_alert import ALERT_TYPE, KEYS, blindspot_alert +from openpilot.selfdrive.selfdrived.alertmanager import AlertManager +from openpilot.selfdrive.selfdrived.events import Alert, AlertSize, AlertStatus, AudibleAlert, ET, Priority, VisualAlert +from openpilot.common.params import Params +from openpilot.starpilot.ui.sounds_owner import SoundsOwner +from openpilot.starpilot.ui.feature_settings_state import FeatureSettingsRequest + +NOW = 10_000_000_000 + + +class Messages(dict): + def __init__(self, active=False): + super().__init__(deviceState=NS(started=True), carControl=NS(latActive=active)) + self.logMonoTime = dict.fromkeys(('deviceState', 'carControl', 'carState'), NOW - 10_000_000) + self.healthy = True + + def all_checks(self, _services): + return self.healthy + + +def car(**kw): + return NS(**(dict(leftBlinker=True, rightBlinker=False, leftBlindspot=True, rightBlindspot=False, + canValid=True, canTimeout=False) | kw)) + + +def test_independent_switches_and_active_lane_change(): + sm = Messages(active=True) + assert blindspot_alert(car(), sm, loud=True, without_lateral=False, lane_change_blocked=True, now_ns=NOW) + assert not blindspot_alert(car(), sm, loud=False, without_lateral=True, lane_change_blocked=False, now_ns=NOW) + sm['carControl'].latActive = False + assert blindspot_alert(car(), sm, loud=False, without_lateral=True, lane_change_blocked=False, now_ns=NOW) + assert blindspot_alert(car(), sm, loud=False, without_lateral=True, lane_change_blocked=True, now_ns=NOW) + assert not blindspot_alert(car(), sm, loud=True, without_lateral=False, lane_change_blocked=False, now_ns=NOW) + assert not blindspot_alert(car(), sm, loud=False, without_lateral=False, lane_change_blocked=True, now_ns=NOW) + + +@pytest.mark.parametrize('fields', [dict(leftBlinker=False), dict(rightBlinker=True), + dict(leftBlindspot=False, rightBlindspot=True), dict(canValid=False), dict(canTimeout=True)]) +def test_no_wrong_side_hazards_or_invalid_can(fields): + assert not blindspot_alert(car(**fields), Messages(), loud=False, without_lateral=True, + lane_change_blocked=False, now_ns=NOW) + + +def test_right_side_and_stale_or_offroad_inputs(): + cs = car(leftBlinker=False, leftBlindspot=False, rightBlinker=True, rightBlindspot=True) + sm = Messages() + def alert(): + return blindspot_alert(cs, sm, loud=False, without_lateral=True, lane_change_blocked=False, now_ns=NOW) + assert alert() + for key in sm.logMonoTime: + stamp = sm.logMonoTime[key] + sm.logMonoTime[key] = NOW - 3_000_000_000 + assert not alert() + sm.logMonoTime[key] = NOW + 1 + assert not alert() + sm.logMonoTime[key] = stamp + sm.healthy = False + assert not alert() + sm.healthy = True + sm['deviceState'].started = False + assert not alert() + + +def test_disengaged_alert_survives_warning_filter_but_never_masks_takeover(): + alert = blindspot_alert(car(), Messages(), loud=False, without_lateral=True, lane_change_blocked=False, now_ns=NOW) + assert alert.audible_alert == AudibleAlert.warningSoft + manager = AlertManager() + manager.add_many(1, [alert]) + manager.process_alerts(1, {ET.WARNING}) + assert manager.current_alert.alert_type == ALERT_TYPE + takeover = Alert('Take Control', '', AlertStatus.critical, AlertSize.full, Priority.HIGHEST, + VisualAlert.steerRequired, AudibleAlert.warningImmediate, 1.) + takeover.alert_type, takeover.event_type = 'takeover', ET.IMMEDIATE_DISABLE + manager.add_many(2, [takeover]) + manager.process_alerts(2, {ET.WARNING}) + assert manager.current_alert.alert_type == 'takeover' + + +def test_switch_defaults_persistence_and_exact_authority(tmp_path): + params = Params(str(tmp_path)) + allowed = True + owner = SoundsOwner(params, lambda: allowed) + for key in KEYS: + row = next(row for row in owner.snapshot().rows if row.key == key) + assert row.value == 'Off' and row.default_value == 'Off' + request = FeatureSettingsRequest(key, row.source, 'On') + assert owner.apply(request) + assert params.get_bool(key) + assert not owner.apply(request) + allowed = False + assert not owner.apply(FeatureSettingsRequest(key, b'1', 'Off')) + allowed = True + assert owner.apply(FeatureSettingsRequest(key, b'1', 'Off')) + + +def test_galaxy_sounds_switches_save_and_reset_with_session_checks(tmp_path): + from openpilot.starpilot.galaxy.settings import AuthorityContext, SettingsGateway, SettingsChanged + params = Params(str(tmp_path)) + context = NS(sample=lambda: AuthorityContext(True, None, None)) + gateway = SettingsGateway(params, context) + for key, label in zip(KEYS, ('Loud Blindspot Alert', 'Blindspot Alert While Lateral Is Off'), strict=True): + page = gateway.page('sounds', 'session', b'generation') + index, row = next((i, row) for i, row in enumerate(page['rows']) if row['label'] == label) + assert row['choices'] == ['Off', 'On'] and row['defaultValue'] == 'Off' + pending = gateway.preview(page['view'], index, 0, 'session', b'generation', value='On') + assert gateway.confirm(pending['intent'], 'session', b'generation') + assert params.get_bool(key) + page = gateway.page('sounds', 'session', b'generation') + reset = gateway.preview(page['view'], index, 0, 'session', b'generation', reset_default=True) + with pytest.raises(SettingsChanged): + gateway.confirm(reset['intent'], 'session', b'generation', session_valid=lambda: False) + assert params.get_bool(key) + reset = gateway.preview(page['view'], index, 0, 'session', b'generation', reset_default=True) + assert gateway.confirm(reset['intent'], 'session', b'generation') + assert not params.get_bool(key) diff --git a/openpilot/starpilot/ui/sounds_owner.py b/openpilot/starpilot/ui/sounds_owner.py index 796fb89672..e4e7129eb5 100644 --- a/openpilot/starpilot/ui/sounds_owner.py +++ b/openpilot/starpilot/ui/sounds_owner.py @@ -7,6 +7,8 @@ from pathlib import Path from openpilot.starpilot.audio.sound_pack import DEFAULT_PACK, PACK_ROOT, installed_packs, read_selection from openpilot.starpilot.saved_document import commit_exact +from openpilot.starpilot.saved_source import read_saved +from openpilot.starpilot.audio.blindspot_alert import KEYS as BLINDSPOT_KEYS from openpilot.starpilot.audio.alert_volume import AUTO, MAX_RAW_BYTES, SPECS, VOLUMES, read_volume from openpilot.starpilot.ui.feature_settings_state import FeatureRow, FeatureSettingsRequest, FeatureSettingsState @@ -57,6 +59,15 @@ class SoundsOwner: rows.append(FeatureRow(AUTO_PREFIX + key, "Use Auto " + label, "Follow ambient sound", source=saved.raw, available=parked and saved.readable, repair_value="Auto")) rows.append(pack_row) + for key, label, reason in ( + (BLINDSPOT_KEYS[0], "Loud Blindspot Alert", "Uses the soft-warning sound when a lane change is blocked by a detected blind-spot vehicle."), + (BLINDSPOT_KEYS[1], "Blindspot Alert While Lateral Is Off", "Warns when signaling toward a detected blind-spot vehicle with steering assistance off or paused. Independent of Loud Blindspot Alert. Hazards do not trigger it."), + ): + raw, readable = read_saved(self.params, key, 8) + valid = readable and raw in (None, b"0", b"1") + rows.append(FeatureRow(key, label, ("On" if raw == b"1" else "Off") if valid else "Invalid saved value", + raw, ("Off", "On") if valid else (), available=parked and valid, + reason=reason, default_value="Off")) defaults = dict.fromkeys(SPECS, 'Auto') defaults['SoundPack'] = PACK_LABELS.get(DEFAULT_PACK, DEFAULT_PACK) rows = tuple(replace(row, default_value=defaults[row.key]) if row.key in defaults else row for row in rows) @@ -65,6 +76,13 @@ class SoundsOwner: rows=tuple(rows), parked=parked) def apply(self, request: FeatureSettingsRequest) -> bool: + if request.key in BLINDSPOT_KEYS: + if request.value not in ("Off", "On") or request.expected not in (None, b"0", b"1"): + return False + result = commit_exact(self.params, key=request.key, max_bytes=8, + raw=b"1" if request.value == "On" else b"0", expected=request.expected, + authorized=self.parked, temp_prefix=".blindspot-alert-") + return result.committed and result.verified if request.key.startswith(AUTO_PREFIX): if request.value != "Auto": return False diff --git a/openpilot/starpilot/ui/tests/test_sounds_settings.py b/openpilot/starpilot/ui/tests/test_sounds_settings.py index ec101a3ef3..b31b96d309 100644 --- a/openpilot/starpilot/ui/tests/test_sounds_settings.py +++ b/openpilot/starpilot/ui/tests/test_sounds_settings.py @@ -49,7 +49,7 @@ class SoundsSettingsTests(unittest.TestCase): def test_absence_auto_no_writes_and_native_registry(self): state = self.owner.snapshot() - self.assertEqual(len(state.rows), 8) + self.assertEqual(len(state.rows), 10) for key, _, _ in VOLUMES: self.assertEqual(self.row(key).value, "Auto") self.assertEqual((self.row(key).step, self.row(key).maximum, self.row(key).unit), (5.0, 100.0, "%")) @@ -264,7 +264,7 @@ class SoundsSettingsTests(unittest.TestCase): session.sounds_scroll = 5 state = replace(session.sounds_snapshot(), scroll=5) self.assertEqual(state.page, "sounds:overview") - self.assertEqual(len(state.rows), 8) + self.assertEqual(len(state.rows), 10) self.assertTrue(all(value_buttons(state, row) == () for row in state.rows)) self.assertEqual(next(row for row in state.rows if row.key == "EngageVolume").value, "40") received = []