You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
983 lines
36 KiB
983 lines
36 KiB
#!/usr/bin/env python3
|
|
import bisect
|
|
import math
|
|
import os
|
|
from enum import IntEnum
|
|
from collections.abc import Callable
|
|
|
|
from cereal import log, car
|
|
import cereal.messaging as messaging
|
|
from openpilot.common.conversions import Conversions as CV
|
|
from openpilot.common.git import get_short_branch
|
|
from openpilot.common.realtime import DT_CTRL
|
|
from openpilot.selfdrive.locationd.calibrationd import MIN_SPEED_FILTER
|
|
|
|
AlertSize = log.SelfdriveState.AlertSize
|
|
AlertStatus = log.SelfdriveState.AlertStatus
|
|
VisualAlert = car.CarControl.HUDControl.VisualAlert
|
|
AudibleAlert = car.CarControl.HUDControl.AudibleAlert
|
|
EventName = log.OnroadEvent.EventName
|
|
|
|
|
|
# Alert priorities
|
|
class Priority(IntEnum):
|
|
LOWEST = 0
|
|
LOWER = 1
|
|
LOW = 2
|
|
MID = 3
|
|
HIGH = 4
|
|
HIGHEST = 5
|
|
|
|
|
|
# Event types
|
|
class ET:
|
|
ENABLE = 'enable'
|
|
PRE_ENABLE = 'preEnable'
|
|
OVERRIDE_LATERAL = 'overrideLateral'
|
|
OVERRIDE_LONGITUDINAL = 'overrideLongitudinal'
|
|
NO_ENTRY = 'noEntry'
|
|
WARNING = 'warning'
|
|
USER_DISABLE = 'userDisable'
|
|
SOFT_DISABLE = 'softDisable'
|
|
IMMEDIATE_DISABLE = 'immediateDisable'
|
|
PERMANENT = 'permanent'
|
|
|
|
|
|
# get event name from enum
|
|
EVENT_NAME = {v: k for k, v in EventName.schema.enumerants.items()}
|
|
|
|
|
|
class Events:
|
|
def __init__(self):
|
|
self.events: list[int] = []
|
|
self.static_events: list[int] = []
|
|
self.event_counters = dict.fromkeys(EVENTS.keys(), 0)
|
|
|
|
@property
|
|
def names(self) -> list[int]:
|
|
return self.events
|
|
|
|
def __len__(self) -> int:
|
|
return len(self.events)
|
|
|
|
def add(self, event_name: int, static: bool=False) -> None:
|
|
if static:
|
|
bisect.insort(self.static_events, event_name)
|
|
bisect.insort(self.events, event_name)
|
|
|
|
def clear(self) -> None:
|
|
self.event_counters = {k: (v + 1 if k in self.events else 0) for k, v in self.event_counters.items()}
|
|
self.events = self.static_events.copy()
|
|
|
|
def contains(self, event_type: str) -> bool:
|
|
return any(event_type in EVENTS.get(e, {}) for e in self.events)
|
|
|
|
def create_alerts(self, event_types: list[str], callback_args=None):
|
|
if callback_args is None:
|
|
callback_args = []
|
|
|
|
ret = []
|
|
for e in self.events:
|
|
types = EVENTS[e].keys()
|
|
for et in event_types:
|
|
if et in types:
|
|
alert = EVENTS[e][et]
|
|
if not isinstance(alert, Alert):
|
|
alert = alert(*callback_args)
|
|
|
|
if DT_CTRL * (self.event_counters[e] + 1) >= alert.creation_delay:
|
|
alert.alert_type = f"{EVENT_NAME[e]}/{et}"
|
|
alert.event_type = et
|
|
ret.append(alert)
|
|
return ret
|
|
|
|
def add_from_msg(self, events):
|
|
for e in events:
|
|
bisect.insort(self.events, e.name.raw)
|
|
|
|
def to_msg(self):
|
|
ret = []
|
|
for event_name in self.events:
|
|
event = log.OnroadEvent.new_message()
|
|
event.name = event_name
|
|
for event_type in EVENTS.get(event_name, {}):
|
|
setattr(event, event_type, True)
|
|
ret.append(event)
|
|
return ret
|
|
|
|
|
|
class Alert:
|
|
def __init__(self,
|
|
alert_text_1: str,
|
|
alert_text_2: str,
|
|
alert_status: log.SelfdriveState.AlertStatus,
|
|
alert_size: log.SelfdriveState.AlertSize,
|
|
priority: Priority,
|
|
visual_alert: car.CarControl.HUDControl.VisualAlert,
|
|
audible_alert: car.CarControl.HUDControl.AudibleAlert,
|
|
duration: float,
|
|
creation_delay: float = 0.):
|
|
|
|
self.alert_text_1 = alert_text_1
|
|
self.alert_text_2 = alert_text_2
|
|
self.alert_status = alert_status
|
|
self.alert_size = alert_size
|
|
self.priority = priority
|
|
self.visual_alert = visual_alert
|
|
self.audible_alert = audible_alert
|
|
|
|
self.duration = int(duration / DT_CTRL)
|
|
|
|
self.creation_delay = creation_delay
|
|
|
|
self.alert_type = ""
|
|
self.event_type: str | None = None
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.alert_text_1}/{self.alert_text_2} {self.priority} {self.visual_alert} {self.audible_alert}"
|
|
|
|
def __gt__(self, alert2) -> bool:
|
|
if not isinstance(alert2, Alert):
|
|
return False
|
|
return self.priority > alert2.priority
|
|
|
|
EmptyAlert = Alert("" , "", AlertStatus.normal, AlertSize.none, Priority.LOWEST,
|
|
VisualAlert.none, AudibleAlert.none, 0)
|
|
|
|
class NoEntryAlert(Alert):
|
|
def __init__(self, alert_text_2: str,
|
|
alert_text_1: str = "openpilot Unavailable",
|
|
visual_alert: car.CarControl.HUDControl.VisualAlert=VisualAlert.none):
|
|
super().__init__(alert_text_1, alert_text_2, AlertStatus.normal,
|
|
AlertSize.mid, Priority.LOW, visual_alert,
|
|
AudibleAlert.refuse, 3.)
|
|
|
|
|
|
class SoftDisableAlert(Alert):
|
|
def __init__(self, alert_text_2: str):
|
|
super().__init__("TAKE CONTROL IMMEDIATELY", alert_text_2,
|
|
AlertStatus.userPrompt, AlertSize.full,
|
|
Priority.MID, VisualAlert.steerRequired,
|
|
AudibleAlert.warningSoft, 2.),
|
|
|
|
|
|
# less harsh version of SoftDisable, where the condition is user-triggered
|
|
class UserSoftDisableAlert(SoftDisableAlert):
|
|
def __init__(self, alert_text_2: str):
|
|
super().__init__(alert_text_2),
|
|
self.alert_text_1 = "openpilot will disengage"
|
|
|
|
|
|
class ImmediateDisableAlert(Alert):
|
|
def __init__(self, alert_text_2: str):
|
|
super().__init__("TAKE CONTROL IMMEDIATELY", alert_text_2,
|
|
AlertStatus.critical, AlertSize.full,
|
|
Priority.HIGHEST, VisualAlert.steerRequired,
|
|
AudibleAlert.warningImmediate, 4.),
|
|
|
|
|
|
class EngagementAlert(Alert):
|
|
def __init__(self, audible_alert: car.CarControl.HUDControl.AudibleAlert):
|
|
super().__init__("", "",
|
|
AlertStatus.normal, AlertSize.none,
|
|
Priority.MID, VisualAlert.none,
|
|
audible_alert, .2),
|
|
|
|
|
|
class NormalPermanentAlert(Alert):
|
|
def __init__(self, alert_text_1: str, alert_text_2: str = "", duration: float = 0.2, priority: Priority = Priority.LOWER, creation_delay: float = 0.):
|
|
super().__init__(alert_text_1, alert_text_2,
|
|
AlertStatus.normal, AlertSize.mid if len(alert_text_2) else AlertSize.small,
|
|
priority, VisualAlert.none, AudibleAlert.none, duration, creation_delay=creation_delay),
|
|
|
|
|
|
class StartupAlert(Alert):
|
|
def __init__(self, alert_text_1: str, alert_text_2: str = "Always keep hands on wheel and eyes on road", alert_status=AlertStatus.normal):
|
|
super().__init__(alert_text_1, alert_text_2,
|
|
alert_status, AlertSize.mid,
|
|
Priority.LOWER, VisualAlert.none, AudibleAlert.none, 5.),
|
|
|
|
|
|
# ********** helper functions **********
|
|
def get_display_speed(speed_ms: float, metric: bool) -> str:
|
|
speed = int(round(speed_ms * (CV.MS_TO_KPH if metric else CV.MS_TO_MPH)))
|
|
unit = 'km/h' if metric else 'mph'
|
|
return f"{speed} {unit}"
|
|
|
|
|
|
# ********** alert callback functions **********
|
|
|
|
AlertCallbackType = Callable[[car.CarParams, car.CarState, messaging.SubMaster, bool, int, log.ControlsState], Alert]
|
|
|
|
|
|
def soft_disable_alert(alert_text_2: str) -> AlertCallbackType:
|
|
def func(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
if soft_disable_time < int(0.5 / DT_CTRL):
|
|
return ImmediateDisableAlert(alert_text_2)
|
|
return SoftDisableAlert(alert_text_2)
|
|
return func
|
|
|
|
def user_soft_disable_alert(alert_text_2: str) -> AlertCallbackType:
|
|
def func(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
if soft_disable_time < int(0.5 / DT_CTRL):
|
|
return ImmediateDisableAlert(alert_text_2)
|
|
return UserSoftDisableAlert(alert_text_2)
|
|
return func
|
|
|
|
def startup_master_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
branch = get_short_branch() # Ensure get_short_branch is cached to avoid lags on startup
|
|
if "REPLAY" in os.environ:
|
|
branch = "replay"
|
|
|
|
return StartupAlert("WARNING: This branch is not tested", branch, alert_status=AlertStatus.userPrompt)
|
|
|
|
def below_engage_speed_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
return NoEntryAlert(f"Drive above {get_display_speed(CP.minEnableSpeed, metric)} to engage")
|
|
|
|
|
|
def below_steer_speed_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
return Alert(
|
|
f"Steer Unavailable Below {get_display_speed(CP.minSteerSpeed, metric)}",
|
|
"",
|
|
AlertStatus.userPrompt, AlertSize.small,
|
|
Priority.LOW, VisualAlert.steerRequired, AudibleAlert.prompt, 0.4)
|
|
|
|
|
|
def calibration_incomplete_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
first_word = 'Recalibration' if sm['liveCalibration'].calStatus == log.LiveCalibrationData.Status.recalibrating else 'Calibration'
|
|
return Alert(
|
|
f"{first_word} in Progress: {sm['liveCalibration'].calPerc:.0f}%",
|
|
f"Drive Above {get_display_speed(MIN_SPEED_FILTER, metric)}",
|
|
AlertStatus.normal, AlertSize.mid,
|
|
Priority.LOWEST, VisualAlert.none, AudibleAlert.none, .2)
|
|
|
|
|
|
# *** debug alerts ***
|
|
|
|
def out_of_space_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
full_perc = round(100. - sm['deviceState'].freeSpacePercent)
|
|
return NormalPermanentAlert("Out of Storage", f"{full_perc}% full")
|
|
|
|
|
|
def posenet_invalid_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
mdl = sm['modelV2'].velocity.x[0] if len(sm['modelV2'].velocity.x) else math.nan
|
|
err = CS.vEgo - mdl
|
|
msg = f"Speed Error: {err:.1f} m/s"
|
|
return NoEntryAlert(msg, alert_text_1="Posenet Speed Invalid")
|
|
|
|
|
|
def process_not_running_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
not_running = [p.name for p in sm['managerState'].processes if not p.running and p.shouldBeRunning]
|
|
msg = ', '.join(not_running)
|
|
return NoEntryAlert(msg, alert_text_1="Process Not Running")
|
|
|
|
|
|
def comm_issue_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
bs = [s for s in sm.data.keys() if not sm.all_checks([s, ])]
|
|
msg = ', '.join(bs[:4]) # can't fit too many on one line
|
|
return NoEntryAlert(msg, alert_text_1="Communication Issue Between Processes")
|
|
|
|
|
|
def camera_malfunction_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
all_cams = ('roadCameraState', 'driverCameraState', 'wideRoadCameraState')
|
|
bad_cams = [s.replace('State', '') for s in all_cams if s in sm.data.keys() and not sm.all_checks([s, ])]
|
|
return NormalPermanentAlert("Camera Malfunction", ', '.join(bad_cams))
|
|
|
|
|
|
def calibration_invalid_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
rpy = sm['liveCalibration'].rpyCalib
|
|
yaw = math.degrees(rpy[2] if len(rpy) == 3 else math.nan)
|
|
pitch = math.degrees(rpy[1] if len(rpy) == 3 else math.nan)
|
|
angles = f"Remount Device (Pitch: {pitch:.1f}°, Yaw: {yaw:.1f}°)"
|
|
return NormalPermanentAlert("Calibration Invalid", angles)
|
|
|
|
|
|
def overheat_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
cpu = max(sm['deviceState'].cpuTempC, default=0.)
|
|
gpu = max(sm['deviceState'].gpuTempC, default=0.)
|
|
temp = max((cpu, gpu, sm['deviceState'].memoryTempC))
|
|
return NormalPermanentAlert("System Overheated", f"{temp:.0f} °C")
|
|
|
|
|
|
def low_memory_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
return NormalPermanentAlert("Low Memory", f"{sm['deviceState'].memoryUsagePercent}% used")
|
|
|
|
|
|
def high_cpu_usage_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
x = max(sm['deviceState'].cpuUsagePercent, default=0.)
|
|
return NormalPermanentAlert("High CPU Usage", f"{x}% used")
|
|
|
|
|
|
def modeld_lagging_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
return NormalPermanentAlert("Driving Model Lagging", f"{sm['modelV2'].frameDropPerc:.1f}% frames dropped")
|
|
|
|
|
|
def wrong_car_mode_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
text = "Enable Adaptive Cruise to Engage"
|
|
if CP.carName == "honda":
|
|
text = "Enable Main Switch to Engage"
|
|
return NoEntryAlert(text)
|
|
|
|
|
|
def joystick_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
gb = sm['carControl'].actuators.accel / 4.
|
|
steer = sm['carControl'].actuators.steer
|
|
vals = f"Gas: {round(gb * 100.)}%, Steer: {round(steer * 100.)}%"
|
|
return NormalPermanentAlert("Joystick Mode", vals)
|
|
|
|
|
|
def longitudinal_maneuver_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
ad = sm['alertDebug']
|
|
audible_alert = AudibleAlert.prompt if 'Active' in ad.alertText1 else AudibleAlert.none
|
|
alert_status = AlertStatus.userPrompt if 'Active' in ad.alertText1 else AlertStatus.normal
|
|
alert_size = AlertSize.mid if ad.alertText2 else AlertSize.small
|
|
return Alert(ad.alertText1, ad.alertText2,
|
|
alert_status, alert_size,
|
|
Priority.LOW, VisualAlert.none, audible_alert, 0.2)
|
|
|
|
|
|
def personality_changed_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality) -> Alert:
|
|
personality = str(personality).title()
|
|
return NormalPermanentAlert(f"Driving Personality: {personality}", duration=1.5)
|
|
|
|
|
|
|
|
EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
|
|
# ********** events with no alerts **********
|
|
|
|
EventName.stockFcw: {},
|
|
EventName.actuatorsApiUnavailable: {},
|
|
|
|
# ********** events only containing alerts displayed in all states **********
|
|
|
|
EventName.joystickDebug: {
|
|
ET.WARNING: joystick_alert,
|
|
ET.PERMANENT: NormalPermanentAlert("Joystick Mode"),
|
|
},
|
|
|
|
EventName.longitudinalManeuver: {
|
|
ET.WARNING: longitudinal_maneuver_alert,
|
|
ET.PERMANENT: NormalPermanentAlert("Longitudinal Maneuver Mode",
|
|
"Ensure road ahead is clear"),
|
|
},
|
|
|
|
EventName.selfdriveInitializing: {
|
|
ET.NO_ENTRY: NoEntryAlert("System Initializing"),
|
|
},
|
|
|
|
EventName.startup: {
|
|
ET.PERMANENT: StartupAlert("Be ready to take over at any time")
|
|
},
|
|
|
|
EventName.startupMaster: {
|
|
ET.PERMANENT: startup_master_alert,
|
|
},
|
|
|
|
EventName.startupNoControl: {
|
|
ET.PERMANENT: StartupAlert("Dashcam mode"),
|
|
ET.NO_ENTRY: NoEntryAlert("Dashcam mode"),
|
|
},
|
|
|
|
EventName.startupNoCar: {
|
|
ET.PERMANENT: StartupAlert("Dashcam mode for unsupported car"),
|
|
},
|
|
|
|
EventName.startupNoSecOcKey: {
|
|
ET.PERMANENT: NormalPermanentAlert("Dashcam Mode",
|
|
"Security Key Not Available",
|
|
priority=Priority.HIGH),
|
|
},
|
|
|
|
EventName.dashcamMode: {
|
|
ET.PERMANENT: NormalPermanentAlert("Dashcam Mode",
|
|
priority=Priority.LOWEST),
|
|
},
|
|
|
|
EventName.invalidLkasSetting: {
|
|
ET.PERMANENT: NormalPermanentAlert("Invalid LKAS setting",
|
|
"Toggle stock LKAS on or off to engage"),
|
|
ET.NO_ENTRY: NoEntryAlert("Invalid LKAS setting"),
|
|
},
|
|
|
|
EventName.cruiseMismatch: {
|
|
#ET.PERMANENT: ImmediateDisableAlert("openpilot failed to cancel cruise"),
|
|
},
|
|
|
|
# openpilot doesn't recognize the car. This switches openpilot into a
|
|
# read-only mode. This can be solved by adding your fingerprint.
|
|
# See https://github.com/commaai/openpilot/wiki/Fingerprinting for more information
|
|
EventName.carUnrecognized: {
|
|
ET.PERMANENT: NormalPermanentAlert("Dashcam Mode",
|
|
"Car Unrecognized",
|
|
priority=Priority.LOWEST),
|
|
},
|
|
|
|
EventName.aeb: {
|
|
ET.PERMANENT: Alert(
|
|
"BRAKE!",
|
|
"Emergency Braking: Risk of Collision",
|
|
AlertStatus.critical, AlertSize.full,
|
|
Priority.HIGHEST, VisualAlert.fcw, AudibleAlert.none, 2.),
|
|
ET.NO_ENTRY: NoEntryAlert("AEB: Risk of Collision"),
|
|
},
|
|
|
|
EventName.stockAeb: {
|
|
ET.PERMANENT: Alert(
|
|
"BRAKE!",
|
|
"Stock AEB: Risk of Collision",
|
|
AlertStatus.critical, AlertSize.full,
|
|
Priority.HIGHEST, VisualAlert.fcw, AudibleAlert.none, 2.),
|
|
ET.NO_ENTRY: NoEntryAlert("Stock AEB: Risk of Collision"),
|
|
},
|
|
|
|
EventName.fcw: {
|
|
ET.PERMANENT: Alert(
|
|
"BRAKE!",
|
|
"Risk of Collision",
|
|
AlertStatus.critical, AlertSize.full,
|
|
Priority.HIGHEST, VisualAlert.fcw, AudibleAlert.warningSoft, 2.),
|
|
},
|
|
|
|
EventName.ldw: {
|
|
ET.PERMANENT: Alert(
|
|
"Lane Departure Detected",
|
|
"",
|
|
AlertStatus.userPrompt, AlertSize.small,
|
|
Priority.LOW, VisualAlert.ldw, AudibleAlert.prompt, 3.),
|
|
},
|
|
|
|
# ********** events only containing alerts that display while engaged **********
|
|
|
|
EventName.steerTempUnavailableSilent: {
|
|
ET.WARNING: Alert(
|
|
"Steering Temporarily Unavailable",
|
|
"",
|
|
AlertStatus.userPrompt, AlertSize.small,
|
|
Priority.LOW, VisualAlert.steerRequired, AudibleAlert.prompt, 1.8),
|
|
},
|
|
|
|
EventName.preDriverDistracted: {
|
|
ET.PERMANENT: Alert(
|
|
"Pay Attention",
|
|
"",
|
|
AlertStatus.normal, AlertSize.small,
|
|
Priority.LOW, VisualAlert.none, AudibleAlert.none, .1),
|
|
},
|
|
|
|
EventName.promptDriverDistracted: {
|
|
ET.PERMANENT: Alert(
|
|
"Pay Attention",
|
|
"Driver Distracted",
|
|
AlertStatus.userPrompt, AlertSize.mid,
|
|
Priority.MID, VisualAlert.steerRequired, AudibleAlert.promptDistracted, .1),
|
|
},
|
|
|
|
EventName.driverDistracted: {
|
|
ET.PERMANENT: Alert(
|
|
"DISENGAGE IMMEDIATELY",
|
|
"Driver Distracted",
|
|
AlertStatus.critical, AlertSize.full,
|
|
Priority.HIGH, VisualAlert.steerRequired, AudibleAlert.warningImmediate, .1),
|
|
},
|
|
|
|
EventName.preDriverUnresponsive: {
|
|
ET.PERMANENT: Alert(
|
|
"Touch Steering Wheel: No Face Detected",
|
|
"",
|
|
AlertStatus.normal, AlertSize.small,
|
|
Priority.LOW, VisualAlert.steerRequired, AudibleAlert.none, .1),
|
|
},
|
|
|
|
EventName.promptDriverUnresponsive: {
|
|
ET.PERMANENT: Alert(
|
|
"Touch Steering Wheel",
|
|
"Driver Unresponsive",
|
|
AlertStatus.userPrompt, AlertSize.mid,
|
|
Priority.MID, VisualAlert.steerRequired, AudibleAlert.promptDistracted, .1),
|
|
},
|
|
|
|
EventName.driverUnresponsive: {
|
|
ET.PERMANENT: Alert(
|
|
"DISENGAGE IMMEDIATELY",
|
|
"Driver Unresponsive",
|
|
AlertStatus.critical, AlertSize.full,
|
|
Priority.HIGH, VisualAlert.steerRequired, AudibleAlert.warningImmediate, .1),
|
|
},
|
|
|
|
EventName.manualRestart: {
|
|
ET.WARNING: Alert(
|
|
"TAKE CONTROL",
|
|
"Resume Driving Manually",
|
|
AlertStatus.userPrompt, AlertSize.mid,
|
|
Priority.LOW, VisualAlert.none, AudibleAlert.none, .2),
|
|
},
|
|
|
|
EventName.resumeRequired: {
|
|
ET.WARNING: Alert(
|
|
"Press Resume to Exit Standstill",
|
|
"",
|
|
AlertStatus.userPrompt, AlertSize.small,
|
|
Priority.LOW, VisualAlert.none, AudibleAlert.none, .2),
|
|
},
|
|
|
|
EventName.belowSteerSpeed: {
|
|
ET.WARNING: below_steer_speed_alert,
|
|
},
|
|
|
|
EventName.preLaneChangeLeft: {
|
|
ET.WARNING: Alert(
|
|
"Steer Left to Start Lane Change Once Safe",
|
|
"",
|
|
AlertStatus.normal, AlertSize.small,
|
|
Priority.LOW, VisualAlert.none, AudibleAlert.none, .1),
|
|
},
|
|
|
|
EventName.preLaneChangeRight: {
|
|
ET.WARNING: Alert(
|
|
"Steer Right to Start Lane Change Once Safe",
|
|
"",
|
|
AlertStatus.normal, AlertSize.small,
|
|
Priority.LOW, VisualAlert.none, AudibleAlert.none, .1),
|
|
},
|
|
|
|
EventName.laneChangeBlocked: {
|
|
ET.WARNING: Alert(
|
|
"Car Detected in Blindspot",
|
|
"",
|
|
AlertStatus.userPrompt, AlertSize.small,
|
|
Priority.LOW, VisualAlert.none, AudibleAlert.prompt, .1),
|
|
},
|
|
|
|
EventName.laneChange: {
|
|
ET.WARNING: Alert(
|
|
"Changing Lanes",
|
|
"",
|
|
AlertStatus.normal, AlertSize.small,
|
|
Priority.LOW, VisualAlert.none, AudibleAlert.none, .1),
|
|
},
|
|
|
|
EventName.steerSaturated: {
|
|
ET.WARNING: Alert(
|
|
"Take Control",
|
|
"Turn Exceeds Steering Limit",
|
|
AlertStatus.userPrompt, AlertSize.mid,
|
|
Priority.LOW, VisualAlert.steerRequired, AudibleAlert.promptRepeat, 2.),
|
|
},
|
|
|
|
# Thrown when the fan is driven at >50% but is not rotating
|
|
EventName.fanMalfunction: {
|
|
ET.PERMANENT: NormalPermanentAlert("Fan Malfunction", "Likely Hardware Issue"),
|
|
},
|
|
|
|
# Camera is not outputting frames
|
|
EventName.cameraMalfunction: {
|
|
ET.PERMANENT: camera_malfunction_alert,
|
|
ET.SOFT_DISABLE: soft_disable_alert("Camera Malfunction"),
|
|
ET.NO_ENTRY: NoEntryAlert("Camera Malfunction: Reboot Your Device"),
|
|
},
|
|
# Camera framerate too low
|
|
EventName.cameraFrameRate: {
|
|
ET.PERMANENT: NormalPermanentAlert("Camera Frame Rate Low", "Reboot your Device"),
|
|
ET.SOFT_DISABLE: soft_disable_alert("Camera Frame Rate Low"),
|
|
ET.NO_ENTRY: NoEntryAlert("Camera Frame Rate Low: Reboot Your Device"),
|
|
},
|
|
|
|
# Unused
|
|
|
|
EventName.locationdTemporaryError: {
|
|
ET.NO_ENTRY: NoEntryAlert("locationd Temporary Error"),
|
|
ET.SOFT_DISABLE: soft_disable_alert("locationd Temporary Error"),
|
|
},
|
|
|
|
EventName.locationdPermanentError: {
|
|
ET.NO_ENTRY: NoEntryAlert("locationd Permanent Error"),
|
|
ET.IMMEDIATE_DISABLE: ImmediateDisableAlert("locationd Permanent Error"),
|
|
ET.PERMANENT: NormalPermanentAlert("locationd Permanent Error"),
|
|
},
|
|
|
|
# openpilot tries to learn certain parameters about your car by observing
|
|
# how the car behaves to steering inputs from both human and openpilot driving.
|
|
# This includes:
|
|
# - steer ratio: gear ratio of the steering rack. Steering angle divided by tire angle
|
|
# - tire stiffness: how much grip your tires have
|
|
# - angle offset: most steering angle sensors are offset and measure a non zero angle when driving straight
|
|
# This alert is thrown when any of these values exceed a sanity check. This can be caused by
|
|
# bad alignment or bad sensor data. If this happens consistently consider creating an issue on GitHub
|
|
EventName.paramsdTemporaryError: {
|
|
ET.NO_ENTRY: NoEntryAlert("paramsd Temporary Error"),
|
|
ET.SOFT_DISABLE: soft_disable_alert("paramsd Temporary Error"),
|
|
},
|
|
|
|
EventName.paramsdPermanentError: {
|
|
ET.NO_ENTRY: NoEntryAlert("paramsd Permanent Error"),
|
|
ET.IMMEDIATE_DISABLE: ImmediateDisableAlert("paramsd Permanent Error"),
|
|
ET.PERMANENT: NormalPermanentAlert("paramsd Permanent Error"),
|
|
},
|
|
|
|
# ********** events that affect controls state transitions **********
|
|
|
|
EventName.pcmEnable: {
|
|
ET.ENABLE: EngagementAlert(AudibleAlert.engage),
|
|
},
|
|
|
|
EventName.buttonEnable: {
|
|
ET.ENABLE: EngagementAlert(AudibleAlert.engage),
|
|
},
|
|
|
|
EventName.pcmDisable: {
|
|
ET.USER_DISABLE: EngagementAlert(AudibleAlert.disengage),
|
|
},
|
|
|
|
EventName.buttonCancel: {
|
|
ET.USER_DISABLE: EngagementAlert(AudibleAlert.disengage),
|
|
ET.NO_ENTRY: NoEntryAlert("Cancel Pressed"),
|
|
},
|
|
|
|
EventName.brakeHold: {
|
|
ET.USER_DISABLE: EngagementAlert(AudibleAlert.disengage),
|
|
ET.NO_ENTRY: NoEntryAlert("Brake Hold Active"),
|
|
},
|
|
|
|
EventName.parkBrake: {
|
|
ET.USER_DISABLE: EngagementAlert(AudibleAlert.disengage),
|
|
ET.NO_ENTRY: NoEntryAlert("Parking Brake Engaged"),
|
|
},
|
|
|
|
EventName.pedalPressed: {
|
|
ET.USER_DISABLE: EngagementAlert(AudibleAlert.disengage),
|
|
ET.NO_ENTRY: NoEntryAlert("Pedal Pressed",
|
|
visual_alert=VisualAlert.brakePressed),
|
|
},
|
|
|
|
EventName.preEnableStandstill: {
|
|
ET.PRE_ENABLE: Alert(
|
|
"Release Brake to Engage",
|
|
"",
|
|
AlertStatus.normal, AlertSize.small,
|
|
Priority.LOWEST, VisualAlert.none, AudibleAlert.none, .1, creation_delay=1.),
|
|
},
|
|
|
|
EventName.gasPressedOverride: {
|
|
ET.OVERRIDE_LONGITUDINAL: Alert(
|
|
"",
|
|
"",
|
|
AlertStatus.normal, AlertSize.none,
|
|
Priority.LOWEST, VisualAlert.none, AudibleAlert.none, .1),
|
|
},
|
|
|
|
EventName.steerOverride: {
|
|
ET.OVERRIDE_LATERAL: Alert(
|
|
"",
|
|
"",
|
|
AlertStatus.normal, AlertSize.none,
|
|
Priority.LOWEST, VisualAlert.none, AudibleAlert.none, .1),
|
|
},
|
|
|
|
EventName.wrongCarMode: {
|
|
ET.USER_DISABLE: EngagementAlert(AudibleAlert.disengage),
|
|
ET.NO_ENTRY: wrong_car_mode_alert,
|
|
},
|
|
|
|
EventName.resumeBlocked: {
|
|
ET.NO_ENTRY: NoEntryAlert("Press Set to Engage"),
|
|
},
|
|
|
|
EventName.wrongCruiseMode: {
|
|
ET.USER_DISABLE: EngagementAlert(AudibleAlert.disengage),
|
|
ET.NO_ENTRY: NoEntryAlert("Adaptive Cruise Disabled"),
|
|
},
|
|
|
|
EventName.steerTempUnavailable: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Steering Temporarily Unavailable"),
|
|
ET.NO_ENTRY: NoEntryAlert("Steering Temporarily Unavailable"),
|
|
},
|
|
|
|
EventName.steerTimeLimit: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Vehicle Steering Time Limit"),
|
|
ET.NO_ENTRY: NoEntryAlert("Vehicle Steering Time Limit"),
|
|
},
|
|
|
|
EventName.outOfSpace: {
|
|
ET.PERMANENT: out_of_space_alert,
|
|
ET.NO_ENTRY: NoEntryAlert("Out of Storage"),
|
|
},
|
|
|
|
EventName.belowEngageSpeed: {
|
|
ET.NO_ENTRY: below_engage_speed_alert,
|
|
},
|
|
|
|
EventName.sensorDataInvalid: {
|
|
ET.PERMANENT: Alert(
|
|
"Sensor Data Invalid",
|
|
"Possible Hardware Issue",
|
|
AlertStatus.normal, AlertSize.mid,
|
|
Priority.LOWER, VisualAlert.none, AudibleAlert.none, .2, creation_delay=1.),
|
|
ET.NO_ENTRY: NoEntryAlert("Sensor Data Invalid"),
|
|
ET.SOFT_DISABLE: soft_disable_alert("Sensor Data Invalid"),
|
|
},
|
|
|
|
EventName.noGps: {
|
|
ET.PERMANENT: Alert(
|
|
"Poor GPS reception",
|
|
"Ensure device has a clear view of the sky",
|
|
AlertStatus.normal, AlertSize.mid,
|
|
Priority.LOWER, VisualAlert.none, AudibleAlert.none, .2, creation_delay=600.)
|
|
},
|
|
|
|
EventName.tooDistracted: {
|
|
ET.NO_ENTRY: NoEntryAlert("Distraction Level Too High"),
|
|
},
|
|
|
|
EventName.overheat: {
|
|
ET.PERMANENT: overheat_alert,
|
|
ET.SOFT_DISABLE: soft_disable_alert("System Overheated"),
|
|
ET.NO_ENTRY: NoEntryAlert("System Overheated"),
|
|
},
|
|
|
|
EventName.wrongGear: {
|
|
ET.SOFT_DISABLE: user_soft_disable_alert("Gear not D"),
|
|
ET.NO_ENTRY: NoEntryAlert("Gear not D"),
|
|
},
|
|
|
|
# This alert is thrown when the calibration angles are outside of the acceptable range.
|
|
# For example if the device is pointed too much to the left or the right.
|
|
# Usually this can only be solved by removing the mount from the windshield completely,
|
|
# and attaching while making sure the device is pointed straight forward and is level.
|
|
# See https://comma.ai/setup for more information
|
|
EventName.calibrationInvalid: {
|
|
ET.PERMANENT: calibration_invalid_alert,
|
|
ET.SOFT_DISABLE: soft_disable_alert("Calibration Invalid: Remount Device & Recalibrate"),
|
|
ET.NO_ENTRY: NoEntryAlert("Calibration Invalid: Remount Device & Recalibrate"),
|
|
},
|
|
|
|
EventName.calibrationIncomplete: {
|
|
ET.PERMANENT: calibration_incomplete_alert,
|
|
ET.SOFT_DISABLE: soft_disable_alert("Calibration Incomplete"),
|
|
ET.NO_ENTRY: NoEntryAlert("Calibration in Progress"),
|
|
},
|
|
|
|
EventName.calibrationRecalibrating: {
|
|
ET.PERMANENT: calibration_incomplete_alert,
|
|
ET.SOFT_DISABLE: soft_disable_alert("Device Remount Detected: Recalibrating"),
|
|
ET.NO_ENTRY: NoEntryAlert("Remount Detected: Recalibrating"),
|
|
},
|
|
|
|
EventName.doorOpen: {
|
|
ET.SOFT_DISABLE: user_soft_disable_alert("Door Open"),
|
|
ET.NO_ENTRY: NoEntryAlert("Door Open"),
|
|
},
|
|
|
|
EventName.seatbeltNotLatched: {
|
|
ET.SOFT_DISABLE: user_soft_disable_alert("Seatbelt Unlatched"),
|
|
ET.NO_ENTRY: NoEntryAlert("Seatbelt Unlatched"),
|
|
},
|
|
|
|
EventName.espDisabled: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Electronic Stability Control Disabled"),
|
|
ET.NO_ENTRY: NoEntryAlert("Electronic Stability Control Disabled"),
|
|
},
|
|
|
|
EventName.lowBattery: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Low Battery"),
|
|
ET.NO_ENTRY: NoEntryAlert("Low Battery"),
|
|
},
|
|
|
|
# Different openpilot services communicate between each other at a certain
|
|
# interval. If communication does not follow the regular schedule this alert
|
|
# is thrown. This can mean a service crashed, did not broadcast a message for
|
|
# ten times the regular interval, or the average interval is more than 10% too high.
|
|
EventName.commIssue: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Communication Issue Between Processes"),
|
|
ET.NO_ENTRY: comm_issue_alert,
|
|
},
|
|
EventName.commIssueAvgFreq: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Low Communication Rate Between Processes"),
|
|
ET.NO_ENTRY: NoEntryAlert("Low Communication Rate Between Processes"),
|
|
},
|
|
|
|
EventName.selfdrivedLagging: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("System Lagging"),
|
|
ET.NO_ENTRY: NoEntryAlert("Selfdrive Process Lagging: Reboot Your Device"),
|
|
},
|
|
|
|
# Thrown when manager detects a service exited unexpectedly while driving
|
|
EventName.processNotRunning: {
|
|
ET.NO_ENTRY: process_not_running_alert,
|
|
ET.SOFT_DISABLE: soft_disable_alert("Process Not Running"),
|
|
},
|
|
|
|
EventName.radarFault: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Radar Error: Restart the Car"),
|
|
ET.NO_ENTRY: NoEntryAlert("Radar Error: Restart the Car"),
|
|
},
|
|
|
|
# Every frame from the camera should be processed by the model. If modeld
|
|
# is not processing frames fast enough they have to be dropped. This alert is
|
|
# thrown when over 20% of frames are dropped.
|
|
EventName.modeldLagging: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Driving Model Lagging"),
|
|
ET.NO_ENTRY: NoEntryAlert("Driving Model Lagging"),
|
|
ET.PERMANENT: modeld_lagging_alert,
|
|
},
|
|
|
|
# Besides predicting the path, lane lines and lead car data the model also
|
|
# predicts the current velocity and rotation speed of the car. If the model is
|
|
# very uncertain about the current velocity while the car is moving, this
|
|
# usually means the model has trouble understanding the scene. This is used
|
|
# as a heuristic to warn the driver.
|
|
EventName.posenetInvalid: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Posenet Speed Invalid"),
|
|
ET.NO_ENTRY: posenet_invalid_alert,
|
|
},
|
|
|
|
# When the localizer detects an acceleration of more than 40 m/s^2 (~4G) we
|
|
# alert the driver the device might have fallen from the windshield.
|
|
EventName.deviceFalling: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Device Fell Off Mount"),
|
|
ET.NO_ENTRY: NoEntryAlert("Device Fell Off Mount"),
|
|
},
|
|
|
|
EventName.lowMemory: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Low Memory: Reboot Your Device"),
|
|
ET.PERMANENT: low_memory_alert,
|
|
ET.NO_ENTRY: NoEntryAlert("Low Memory: Reboot Your Device"),
|
|
},
|
|
|
|
EventName.accFaulted: {
|
|
ET.IMMEDIATE_DISABLE: ImmediateDisableAlert("Cruise Fault: Restart the Car"),
|
|
ET.PERMANENT: NormalPermanentAlert("Cruise Fault: Restart the car to engage"),
|
|
ET.NO_ENTRY: NoEntryAlert("Cruise Fault: Restart the Car"),
|
|
},
|
|
|
|
EventName.espActive: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("Electronic Stability Control Active"),
|
|
ET.NO_ENTRY: NoEntryAlert("Electronic Stability Control Active"),
|
|
},
|
|
|
|
EventName.controlsMismatch: {
|
|
ET.IMMEDIATE_DISABLE: ImmediateDisableAlert("Controls Mismatch"),
|
|
ET.NO_ENTRY: NoEntryAlert("Controls Mismatch"),
|
|
},
|
|
|
|
# Sometimes the USB stack on the device can get into a bad state
|
|
# causing the connection to the panda to be lost
|
|
EventName.usbError: {
|
|
ET.SOFT_DISABLE: soft_disable_alert("USB Error: Reboot Your Device"),
|
|
ET.PERMANENT: NormalPermanentAlert("USB Error: Reboot Your Device", ""),
|
|
ET.NO_ENTRY: NoEntryAlert("USB Error: Reboot Your Device"),
|
|
},
|
|
|
|
# This alert can be thrown for the following reasons:
|
|
# - No CAN data received at all
|
|
# - CAN data is received, but some message are not received at the right frequency
|
|
# If you're not writing a new car port, this is usually cause by faulty wiring
|
|
EventName.canError: {
|
|
ET.IMMEDIATE_DISABLE: ImmediateDisableAlert("CAN Error"),
|
|
ET.PERMANENT: Alert(
|
|
"CAN Error: Check Connections",
|
|
"",
|
|
AlertStatus.normal, AlertSize.small,
|
|
Priority.LOW, VisualAlert.none, AudibleAlert.none, 1., creation_delay=1.),
|
|
ET.NO_ENTRY: NoEntryAlert("CAN Error: Check Connections"),
|
|
},
|
|
|
|
EventName.canBusMissing: {
|
|
ET.IMMEDIATE_DISABLE: ImmediateDisableAlert("CAN Bus Disconnected"),
|
|
ET.PERMANENT: Alert(
|
|
"CAN Bus Disconnected: Likely Faulty Cable",
|
|
"",
|
|
AlertStatus.normal, AlertSize.small,
|
|
Priority.LOW, VisualAlert.none, AudibleAlert.none, 1., creation_delay=1.),
|
|
ET.NO_ENTRY: NoEntryAlert("CAN Bus Disconnected: Check Connections"),
|
|
},
|
|
|
|
EventName.steerUnavailable: {
|
|
ET.IMMEDIATE_DISABLE: ImmediateDisableAlert("LKAS Fault: Restart the Car"),
|
|
ET.PERMANENT: NormalPermanentAlert("LKAS Fault: Restart the car to engage"),
|
|
ET.NO_ENTRY: NoEntryAlert("LKAS Fault: Restart the Car"),
|
|
},
|
|
|
|
EventName.reverseGear: {
|
|
ET.PERMANENT: Alert(
|
|
"Reverse\nGear",
|
|
"",
|
|
AlertStatus.normal, AlertSize.full,
|
|
Priority.LOWEST, VisualAlert.none, AudibleAlert.none, .2, creation_delay=0.5),
|
|
ET.USER_DISABLE: ImmediateDisableAlert("Reverse Gear"),
|
|
ET.NO_ENTRY: NoEntryAlert("Reverse Gear"),
|
|
},
|
|
|
|
# On cars that use stock ACC the car can decide to cancel ACC for various reasons.
|
|
# When this happens we can no long control the car so the user needs to be warned immediately.
|
|
EventName.cruiseDisabled: {
|
|
ET.IMMEDIATE_DISABLE: ImmediateDisableAlert("Cruise Is Off"),
|
|
},
|
|
|
|
# When the relay in the harness box opens the CAN bus between the LKAS camera
|
|
# and the rest of the car is separated. When messages from the LKAS camera
|
|
# are received on the car side this usually means the relay hasn't opened correctly
|
|
# and this alert is thrown.
|
|
EventName.relayMalfunction: {
|
|
ET.IMMEDIATE_DISABLE: ImmediateDisableAlert("Harness Relay Malfunction"),
|
|
ET.PERMANENT: NormalPermanentAlert("Harness Relay Malfunction", "Check Hardware"),
|
|
ET.NO_ENTRY: NoEntryAlert("Harness Relay Malfunction"),
|
|
},
|
|
|
|
EventName.speedTooLow: {
|
|
ET.IMMEDIATE_DISABLE: Alert(
|
|
"openpilot Canceled",
|
|
"Speed too low",
|
|
AlertStatus.normal, AlertSize.mid,
|
|
Priority.HIGH, VisualAlert.none, AudibleAlert.disengage, 3.),
|
|
},
|
|
|
|
# When the car is driving faster than most cars in the training data, the model outputs can be unpredictable.
|
|
EventName.speedTooHigh: {
|
|
ET.WARNING: Alert(
|
|
"Speed Too High",
|
|
"Model uncertain at this speed",
|
|
AlertStatus.userPrompt, AlertSize.mid,
|
|
Priority.HIGH, VisualAlert.steerRequired, AudibleAlert.promptRepeat, 4.),
|
|
ET.NO_ENTRY: NoEntryAlert("Slow down to engage"),
|
|
},
|
|
|
|
EventName.vehicleSensorsInvalid: {
|
|
ET.IMMEDIATE_DISABLE: ImmediateDisableAlert("Vehicle Sensors Invalid"),
|
|
ET.PERMANENT: NormalPermanentAlert("Vehicle Sensors Calibrating", "Drive to Calibrate"),
|
|
ET.NO_ENTRY: NoEntryAlert("Vehicle Sensors Calibrating"),
|
|
},
|
|
|
|
EventName.personalityChanged: {
|
|
ET.WARNING: personality_changed_alert,
|
|
},
|
|
|
|
}
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# print all alerts by type and priority
|
|
from cereal.services import SERVICE_LIST
|
|
from collections import defaultdict
|
|
|
|
event_names = {v: k for k, v in EventName.schema.enumerants.items()}
|
|
alerts_by_type: dict[str, dict[Priority, list[str]]] = defaultdict(lambda: defaultdict(list))
|
|
|
|
CP = car.CarParams.new_message()
|
|
CS = car.CarState.new_message()
|
|
sm = messaging.SubMaster(list(SERVICE_LIST.keys()))
|
|
|
|
for i, alerts in EVENTS.items():
|
|
for et, alert in alerts.items():
|
|
if callable(alert):
|
|
alert = alert(CP, CS, sm, False, 1, log.LongitudinalPersonality.standard)
|
|
alerts_by_type[et][alert.priority].append(event_names[i])
|
|
|
|
all_alerts: dict[str, list[tuple[Priority, list[str]]]] = {}
|
|
for et, priority_alerts in alerts_by_type.items():
|
|
all_alerts[et] = sorted(priority_alerts.items(), key=lambda x: x[0], reverse=True)
|
|
|
|
for status, evs in sorted(all_alerts.items(), key=lambda x: x[0]):
|
|
print(f"**** {status} ****")
|
|
for p, alert_list in evs:
|
|
print(f" {repr(p)}:")
|
|
print(" ", ', '.join(alert_list), "\n")
|
|
|