Blindspot alert

openpilot/common/params_keys.h

@@ -135,6 +135,8 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> 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}"}},

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)

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

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)

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

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 = []