IQ.Pilot Release Commit @ 461be14

iqpilot/selfdrive/selfdrived/events.py

@@ -375,7 +375,7 @@ EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
       "Pay Attention",
       "",
       AlertStatus.normal, AlertSize.small,
-      Priority.LOW, VisualAlert.none, AudibleAlert.none, .1),
+      Priority.MID, VisualAlert.none, AudibleAlert.none, .1),
   },
 
   EventName.promptDriverDistracted: {
@@ -901,7 +901,7 @@ if HARDWARE.get_device_type() == 'mici':
         "Pay Attention",
         "",
         AlertStatus.normal, AlertSize.small,
-        Priority.LOW, VisualAlert.none, AudibleAlert.none, 2),
+        Priority.MID, VisualAlert.none, AudibleAlert.none, 2),
     },
     EventName.promptDriverDistracted: {
       ET.PERMANENT: Alert(

iqpilot/selfdrive/selfdrived/iq_events.py

@@ -1,5 +1,6 @@
 import iqpilot.cereal.messaging as messaging
 from iqpilot.cereal import log, car, custom
+from iqpilot.common.params import Params
 from iqpilot.common.constants import CV
 from iqpilot.common.atlas_alerts import EventBook as EventsBase, Tier as Priority, Tags as ET, AlertCard as Alert, \
   NoEntryCard as NoEntryAlert, HardDisableCard as ImmediateDisableAlert, ChimeCard as EngagementAlert, \
@@ -94,6 +95,24 @@ _CAMERA_LABELS = {
 }
 
 _POLICE_CHIMED_IDS: set[str] = set()
+_USA_REGION_CODES = frozenset(("US", "USA", "UNITED STATES", "UNITED STATES OF AMERICA"))
+
+
+def _configured_country_code() -> str:
+  try:
+    value = Params().get("OsmLocationName")
+  except Exception:
+    return ""
+  if isinstance(value, bytes):
+    value = value.decode("utf-8", "ignore")
+  return str(value or "").strip().upper()
+
+
+def _alpr_alert_labels(country_code: str) -> tuple[str, str]:
+  is_row = bool(country_code) and country_code not in _USA_REGION_CODES
+  if is_row:
+    return "Traffic / ALPR Camera", "Traffic / ALPR Camera Detected"
+  return "Flock / ALPR Camera", "Flock Camera Detected"
 
 
 def speed_camera_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
@@ -101,14 +120,17 @@ def speed_camera_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMas
   ctype = int(getattr(nav.cameraType, "raw", nav.cameraType))
   label = _CAMERA_LABELS.get(ctype, "Speed Camera")
   distance = float(nav.cameraDistance)
+  alpr_detected_label = "Flock Camera Detected"
+  if ctype == int(custom.IQNavState.CameraType.alpr):
+    label, alpr_detected_label = _alpr_alert_labels(_configured_country_code())
   # RF (BLE/WiFi) Flock detection is a live proximity hit with no meaningful
   # distance — flockd/navd flag it with distance 0 on the alpr camera type.
   if ctype == int(custom.IQNavState.CameraType.alpr) and distance <= 0.0:
     return Alert(
-      "Flock Camera Detected",
+      alpr_detected_label,
       "",
       AlertStatus.normal, AlertSize.small,
-      Priority.HIGH, VisualAlert.none, AudibleAlert.prompt, .2)
+      Priority.LOW, VisualAlert.none, AudibleAlert.prompt, .2)
   if metric:
     dist_str = f"{distance:.0f} m" if distance < 1000.0 else f"{distance / 1000.0:.1f} km"
   else:
@@ -134,7 +156,8 @@ def speed_camera_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMas
     f"{label} • {detail}",
     "",
     AlertStatus.normal, AlertSize.small,
-    Priority.HIGH, VisualAlert.none, audible, .2)
+    Priority.LOW if ctype == int(custom.IQNavState.CameraType.alpr) else Priority.HIGH,
+    VisualAlert.none, audible, .2)
 
 
 class IQEvents(EventsBase):

iqpilot/selfdrive/selfdrived/tests/test_iq_camera_alerts.py

@@ -1,13 +1,17 @@
+import copy
 from types import SimpleNamespace
 
-from iqpilot.cereal import car, custom
+from iqpilot.cereal import car, custom, log
+from iqpilot.common.atlas_alerts import HardDisableCard, Tags as ET, Tier as Priority
+from iqpilot.selfdrive.selfdrived.alertmanager import AlertManager
+from iqpilot.selfdrive.selfdrived.events import EVENTS
 from iqpilot.selfdrive.selfdrived import iq_events
 
 
-def alert(camera_type, *, report_id="", chime=False):
+def alert(camera_type, *, report_id="", chime=False, distance=300.0):
   nav = SimpleNamespace(
     cameraType=camera_type,
-    cameraDistance=300.0,
+    cameraDistance=distance,
     cameraSpeedLimit=25.0,
     cameraAlertId=report_id,
     cameraChime=chime,
@@ -31,3 +35,45 @@ def test_police_chime_is_deduplicated_by_report():
   second = alert(custom.IQNavState.CameraType.police, report_id="police-a", chime=True)
   assert first.audible_alert == car.CarControl.HUDControl.AudibleAlert.prompt
   assert second.audible_alert == car.CarControl.HUDControl.AudibleAlert.none
+
+
+def test_alpr_wording_uses_configured_region(monkeypatch):
+  monkeypatch.setattr(iq_events, "_configured_country_code", lambda: "US")
+  assert alert(custom.IQNavState.CameraType.alpr).alert_text_1.startswith("Flock / ALPR Camera")
+  assert alert(custom.IQNavState.CameraType.alpr, distance=0.0).alert_text_1 == "Flock Camera Detected"
+
+  monkeypatch.setattr(iq_events, "_configured_country_code", lambda: "DE")
+  assert alert(custom.IQNavState.CameraType.alpr).alert_text_1.startswith("Traffic / ALPR Camera")
+  assert alert(custom.IQNavState.CameraType.alpr, distance=0.0).alert_text_1 == "Traffic / ALPR Camera Detected"
+
+
+def test_missing_region_is_safe_and_preserves_flock_wording(monkeypatch):
+  class UnavailableParams:
+    def get(self, key):
+      raise OSError(key)
+
+  monkeypatch.setattr(iq_events, "Params", UnavailableParams)
+  assert iq_events._configured_country_code() == ""
+  assert alert(custom.IQNavState.CameraType.alpr, distance=0.0).alert_text_1 == "Flock Camera Detected"
+
+
+def test_driver_attention_and_takeover_alerts_preempt_alpr():
+  flock = alert(custom.IQNavState.CameraType.alpr, distance=0.0)
+  pre_attention = copy.copy(EVENTS[log.OnroadEvent.EventName.preDriverDistracted][ET.PERMANENT])
+  prompt_attention = copy.copy(EVENTS[log.OnroadEvent.EventName.promptDriverDistracted][ET.PERMANENT])
+  takeover = copy.copy(EVENTS[log.OnroadEvent.EventName.driverDistracted][ET.PERMANENT])
+  immediate_disable = HardDisableCard("Regression Test")
+
+  assert flock.priority == Priority.LOW
+  assert pre_attention.priority == flock.priority + 1
+  assert prompt_attention.priority == flock.priority + 1
+  assert takeover.priority > flock.priority
+  assert immediate_disable.priority > flock.priority
+
+  for expected in (pre_attention, prompt_attention, takeover, immediate_disable):
+    manager = AlertManager()
+    flock.alert_type = "flock/warning"
+    expected.alert_type = f"expected/{expected.alert_text_1}"
+    manager.add_many(0, [flock, expected])
+    manager.process_alerts(0, set())
+    assert manager.current_alert is expected