IQ.Lvbs/IQ.Pilot
8
7
Fork 3
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

IQ.Pilot Release Commit @ d23cc80

Browse Source
This commit is contained in:
IQ.Lvbs CI [bot]
2026-07-27 18:01:16 -05:00
parent 46277b19ec
commit a70938008a
74 changed files with 715 additions and 1296 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
selfdrive/selfdrived/events.py
View File

@@ -79,15 +79,6 @@ def below_steer_speed_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.S
Priority.LOW, VisualAlert.none, AudibleAlert.prompt, 0.4)
def too_distracted_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
if sm['driverMonitoringState'].lockout:
mins_left = sm['driverMonitoringState'].lockoutMinutesRemaining
if mins_left <= 0:
return NoEntryAlert("Distraction Level Too High", priority=Priority.HIGH)
return NoEntryAlert("Too Distracted", f"{mins_left} minute{'s' if mins_left != 1 else ''} Left", priority=Priority.HIGH)
return NoEntryAlert("Pay Attention to Engage", priority=Priority.HIGH)
def calibration_incomplete_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
first_word = 'Recalibrating' if sm['liveCalibration'].calStatus == log.LiveCalibrationData.Status.recalibrating else 'Calibrating'
return Alert(
@@ -235,8 +226,7 @@ def invalid_lkas_setting_alert(CP: car.CarParams, CS: car.CarState, sm: messagin
return NormalPermanentAlert(title, text)
def invalid_lkas_setting_no_entry_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster,
metric: bool, soft_disable_time: int, personality) -> Alert:
def invalid_lkas_setting_no_entry_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
if CP.brand == "tesla":
return NoEntryAlert("FSD / Autosteer is active", alert_text_1="Dashcam Mode")
return NoEntryAlert("Invalid LKAS setting")
@@ -370,15 +360,15 @@ EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
Priority.LOW, VisualAlert.steerRequired, AudibleAlert.prompt, 1.8),
},
EventName.driverDistracted1: {
EventName.preDriverDistracted: {
ET.PERMANENT: Alert(
"Pay Attention",
"",
AlertStatus.normal, AlertSize.small,
Priority.LOW, VisualAlert.none, AudibleAlert.preAlert, .1),
Priority.LOW, VisualAlert.none, AudibleAlert.none, .1),
},
EventName.driverDistracted2: {
EventName.promptDriverDistracted: {
ET.PERMANENT: Alert(
"Pay Attention",
"Driver Distracted",
@@ -386,7 +376,7 @@ EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
Priority.MID, VisualAlert.steerRequired, AudibleAlert.promptDistracted, .1),
},
EventName.driverDistracted3: {
EventName.driverDistracted: {
ET.PERMANENT: Alert(
"DISENGAGE IMMEDIATELY",
"Driver Distracted",
@@ -394,7 +384,7 @@ EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
Priority.HIGH, VisualAlert.steerRequired, AudibleAlert.warningImmediate, .1),
},
EventName.driverUnresponsive1: {
EventName.preDriverUnresponsive: {
ET.PERMANENT: Alert(
"Touch Steering Wheel: No Face Detected",
"",
@@ -402,7 +392,7 @@ EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
Priority.LOW, VisualAlert.steerRequired, AudibleAlert.none, .1),
},
EventName.driverUnresponsive2: {
EventName.promptDriverUnresponsive: {
ET.PERMANENT: Alert(
"Touch Steering Wheel",
"Driver Unresponsive",
@@ -410,7 +400,7 @@ EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
Priority.MID, VisualAlert.steerRequired, AudibleAlert.promptDistracted, .1),
},
EventName.driverUnresponsive3: {
EventName.driverUnresponsive: {
ET.PERMANENT: Alert(
"DISENGAGE IMMEDIATELY",
"Driver Unresponsive",
@@ -642,7 +632,7 @@ EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
},
EventName.tooDistracted: {
ET.NO_ENTRY: too_distracted_alert,
ET.NO_ENTRY: NoEntryAlert("Distraction Level Too High"),
},
EventName.excessiveActuation: {
@@ -888,14 +878,14 @@ EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
if HARDWARE.get_device_type() == 'mici':
EVENTS.update({
EventName.driverDistracted1: {
EventName.preDriverDistracted: {
ET.PERMANENT: Alert(
"Pay Attention",
"",
AlertStatus.normal, AlertSize.small,
Priority.LOW, VisualAlert.none, AudibleAlert.preAlert, 2),
Priority.LOW, VisualAlert.none, AudibleAlert.none, 2),
},
EventName.driverDistracted2: {
EventName.promptDriverDistracted: {
ET.PERMANENT: Alert(
"Pay Attention",
"Driver Distracted",

32
selfdrive/selfdrived/selfdrived.py
View File

@@ -54,8 +54,6 @@ EventName = log.OnroadEvent.EventName
ButtonType = car.CarState.ButtonEvent.Type
SafetyModel = car.CarParams.SafetyModel
TurnDirection = custom.IQTurnSignalDirection
AlertLevel = log.DriverMonitoringState.AlertLevel
MonitoringPolicy = log.DriverMonitoringState.MonitoringPolicy
IGNORED_SAFETY_MODES = (SafetyModel.silent, SafetyModel.noOutput)
@@ -129,6 +127,7 @@ class SelfdriveD(GapButtonActions):
self.is_ldw_enabled = self.params.get_bool("IsLdwEnabled")
self.disengage_on_accelerator = self.params.get_bool("DisengageOnAccelerator")
self.nav_exit_lane_change = self._read_nav_exit_lane_change()
self.model_download_pending = self.params.get("ModelManager_DownloadIndex") is not None
car_recognized = self.CP.brand != 'mock'
@@ -149,7 +148,6 @@ class SelfdriveD(GapButtonActions):
self.events_prev = []
self.logged_comm_issue = None
self.not_running_prev = None
self.dm_lockout_set = False
self.experimental_mode = False
self.personality = get_sanitize_int_param(
"LongitudinalPersonality",
@@ -186,6 +184,7 @@ class SelfdriveD(GapButtonActions):
self.events_iq = IQEvents()
self.events_iq_prev = []
self._cached_dm_event_names: tuple[int, ...] = ()
self._cached_plan_event_names: tuple[int, ...] = ()
self._cached_model_event_names: tuple[int, ...] = ()
self._cached_nav_event_names: tuple[int, ...] = ()
@@ -207,6 +206,10 @@ class SelfdriveD(GapButtonActions):
if self.sm.updated['iqPlan']:
self._cached_plan_event_names = tuple(event.name.raw for event in self._get_longitudinal_plan_ext().events)
def _refresh_cached_dm_events(self) -> None:
if self.sm.updated['driverMonitoringState']:
self._cached_dm_event_names = tuple(event.name.raw for event in self.sm['driverMonitoringState'].events)
def _refresh_cached_model_events(self) -> None:
if not self.sm.updated['iqDriveModelData']:
return
@@ -295,24 +298,8 @@ class SelfdriveD(GapButtonActions):
self.events.add(EventName.resumeBlocked)
if not self.CP.notCar:
if self.sm['driverMonitoringState'].lockout and not self.dm_lockout_set:
self.params.put_bool("DriverTooDistracted", True)
self.dm_lockout_set = True
elif not self.sm['driverMonitoringState'].lockout and self.dm_lockout_set:
self.params.remove("DriverTooDistracted")
self.dm_lockout_set = False
if self.sm['driverMonitoringState'].lockout or self.sm['driverMonitoringState'].alwaysOnLockout:
self.events.add(EventName.tooDistracted)
vision_dm = self.sm['driverMonitoringState'].activePolicy == MonitoringPolicy.vision
if self.sm['driverMonitoringState'].alertLevel == AlertLevel.one:
self.events.add(EventName.driverDistracted1 if vision_dm else EventName.driverUnresponsive1)
elif self.sm['driverMonitoringState'].alertLevel == AlertLevel.two:
self.events.add(EventName.driverDistracted2 if vision_dm else EventName.driverUnresponsive2)
elif self.sm['driverMonitoringState'].alertLevel == AlertLevel.three:
self.events.add(EventName.driverDistracted3 if vision_dm else EventName.driverUnresponsive3)
self._refresh_cached_dm_events()
self._add_event_names(self._cached_dm_event_names)
self._refresh_cached_plan_events()
self._add_iq_event_names(self._cached_plan_event_names)
@@ -443,6 +430,8 @@ class SelfdriveD(GapButtonActions):
self.not_running_prev = not_running
if self.sm.recv_frame['managerState'] and (not_running - self.ignored_processes):
self.events.add(EventName.processNotRunning)
if 'iqmodeld' in not_running and self.model_download_pending:
self.events_iq.add(custom.IQOnroadEvent.EventName.modelUpdating)
else:
if not SIMULATION and not self.rk.lagging:
if not self.sm.all_alive(self.camera_packets):
@@ -756,6 +745,7 @@ class SelfdriveD(GapButtonActions):
self.experimental_mode = self.params.get_bool("ExperimentalMode") and self.CP.openpilotLongitudinalControl
self.personality = self.params.get("LongitudinalPersonality", return_default=True)
self.nav_exit_lane_change = self._read_nav_exit_lane_change()
self.model_download_pending = self.params.get("ModelManager_DownloadIndex") is not None
self.aol.read_params()
time.sleep(0.1)

40
selfdrive/selfdrived/tests/test_driver_monitoring.py
View File

@@ -1,40 +0,0 @@
from types import SimpleNamespace
from cereal import car, log
from openpilot.selfdrive.selfdrived.events import EVENTS, ET
EventName = log.OnroadEvent.EventName
AudibleAlert = car.CarControl.HUDControl.AudibleAlert
def test_driver_monitoring_alert_stages():
expected_sounds = {
EventName.driverDistracted1: AudibleAlert.preAlert,
EventName.driverDistracted2: AudibleAlert.promptDistracted,
EventName.driverDistracted3: AudibleAlert.warningImmediate,
EventName.driverUnresponsive1: AudibleAlert.none,
EventName.driverUnresponsive2: AudibleAlert.promptDistracted,
EventName.driverUnresponsive3: AudibleAlert.warningImmediate,
}
for event_name, audible_alert in expected_sounds.items():
assert EVENTS[event_name][ET.PERMANENT].audible_alert == audible_alert
def test_driver_monitoring_lockout_alert():
callback = EVENTS[EventName.tooDistracted][ET.NO_ENTRY]
sm = {'driverMonitoringState': SimpleNamespace(lockout=True, lockoutMinutesRemaining=5)}
alert = callback(None, None, sm, False, 0, None)
assert alert.alert_text_1 == "5 minutes Left"
assert alert.alert_text_2 == "Too Distracted"
def test_legacy_driver_monitoring_lockout_alert():
callback = EVENTS[EventName.tooDistracted][ET.NO_ENTRY]
sm = {'driverMonitoringState': SimpleNamespace(lockout=True, lockoutMinutesRemaining=0)}
alert = callback(None, None, sm, False, 0, None)
assert alert.alert_text_2 == "Distraction Level Too High"