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.
137 lines
8.5 KiB
137 lines
8.5 KiB
from cereal import car
|
|
from selfdrive.swaglog import cloudlog
|
|
import copy
|
|
|
|
class ET:
|
|
ENABLE = 0
|
|
NO_ENTRY = 1
|
|
WARNING = 2
|
|
USER_DISABLE = 3
|
|
SOFT_DISABLE = 4
|
|
IMMEDIATE_DISABLE = 5
|
|
|
|
class alert(object):
|
|
def __init__(self, alert_text_1, alert_text_2, alert_type, visual_alert, audible_alert, duration_sound, duration_hud_alert, duration_text):
|
|
self.alert_text_1 = alert_text_1
|
|
self.alert_text_2 = alert_text_2
|
|
self.alert_type = alert_type
|
|
self.visual_alert = visual_alert if visual_alert is not None else "none"
|
|
self.audible_alert = audible_alert if audible_alert is not None else "none"
|
|
|
|
self.duration_sound = duration_sound
|
|
self.duration_hud_alert = duration_hud_alert
|
|
self.duration_text = duration_text
|
|
|
|
# typecheck that enums are valid on startup
|
|
tst = car.CarControl.new_message()
|
|
tst.hudControl.visualAlert = self.visual_alert
|
|
tst.hudControl.audibleAlert = self.audible_alert
|
|
|
|
def __str__(self):
|
|
return self.alert_text_1 + "/" + self.alert_text_2 + " " + str(self.alert_type) + " " + str(self.visual_alert) + " " + str(self.audible_alert)
|
|
|
|
def __gt__(self, alert2):
|
|
return self.alert_type > alert2.alert_type
|
|
|
|
class AlertManager(object):
|
|
alerts = {
|
|
"enable": alert("", "", ET.ENABLE, None, "beepSingle", .2, 0., 0.),
|
|
"disable": alert("", "", ET.USER_DISABLE, None, "beepSingle", .2, 0., 0.),
|
|
"pedalPressed": alert("Comma Unavailable", "Pedal Pressed", ET.NO_ENTRY, "brakePressed", "chimeDouble", .4, 2., 3.),
|
|
"preDriverDistracted": alert("Take Control ", "User Distracted", ET.WARNING, "steerRequired", "chimeDouble", .4, 2., 3.),
|
|
"driverDistracted": alert("Take Control to Regain Speed", "User Distracted", ET.WARNING, "steerRequired", "chimeRepeated", .5, .5, .5),
|
|
"steerSaturated": alert("Take Control", "Turn Exceeds Limit", ET.WARNING, "steerRequired", "chimeSingle", 1., 2., 3.),
|
|
"overheat": alert("Take Control Immediately", "System Overheated", ET.SOFT_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"controlsMismatch": alert("Take Control Immediately", "Controls Mismatch", ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"radarCommIssue": alert("Take Control Immediately", "Radar Error: Restart the Car",ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"calibrationInvalid": alert("Take Control Immediately", "Calibration Invalid: Reposition Neo and Recalibrate", ET.SOFT_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"calibrationInProgress": alert("Take Control Immediately", "Calibration in Progress: ", ET.SOFT_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"modelCommIssue": alert("Take Control Immediately", "Model Error: Restart the Car",ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"controlsFailed": alert("Take Control Immediately", "Controls Failed", ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"fcw": alert("", "", ET.WARNING, None, None, .1, .1, .1),
|
|
# car errors
|
|
"commIssue": alert("Take Control Immediately","CAN Error: Restart the Car", ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"steerUnavailable": alert("Take Control Immediately","Steer Error: Restart the Car", ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"steerTempUnavailable": alert("Take Control", "Steer Temporarily Unavailable", ET.WARNING, "steerRequired", "chimeDouble", .4, 2., 3.),
|
|
"brakeUnavailable": alert("Take Control Immediately","Brake Error: Restart the Car", ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"gasUnavailable": alert("Take Control Immediately","Gas Error: Restart the Car", ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"wrongGear": alert("Take Control Immediately","Gear not D", ET.SOFT_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"reverseGear": alert("Take Control Immediately","Car in Reverse", ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"doorOpen": alert("Take Control Immediately","Door Open", ET.SOFT_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"seatbeltNotLatched": alert("Take Control Immediately","Seatbelt Unlatched", ET.SOFT_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"espDisabled": alert("Take Control Immediately","ESP Off", ET.SOFT_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"cruiseDisabled": alert("Take Control Immediately","Cruise Is Off", ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"wrongCarMode": alert("Comma Unavailable","Main Switch Off", ET.NO_ENTRY, None, "chimeDouble", .4, 0., 3.),
|
|
"outOfSpace": alert("Comma Unavailable","Out of Space", ET.NO_ENTRY, None, "chimeDouble", .4, 0., 3.),
|
|
"dataNeeded": alert("Comma Unavailable","Data needed for calibration. Upload drive, try again", ET.NO_ENTRY, None, "chimeDouble", .4, 0., 3.),
|
|
"ethicalDilemma": alert("Take Control Immediately","Ethical Dilemma Detected", ET.IMMEDIATE_DISABLE, "steerRequired", "chimeRepeated", 1., 3., 3.),
|
|
"startup": alert("Always Keep Hands on Wheel","Be Ready to Take Over Any Time", ET.NO_ENTRY, None, None, 0., 0., 15.),
|
|
}
|
|
def __init__(self):
|
|
self.activealerts = []
|
|
self.current_alert = None
|
|
self.add("startup", False)
|
|
|
|
def alertPresent(self):
|
|
return len(self.activealerts) > 0
|
|
|
|
def alertShouldSoftDisable(self):
|
|
return len(self.activealerts) > 0 and any(a.alert_type == ET.SOFT_DISABLE for a in self.activealerts)
|
|
|
|
def alertShouldDisable(self):
|
|
return len(self.activealerts) > 0 and any(a.alert_type in [ET.IMMEDIATE_DISABLE, ET.USER_DISABLE] for a in self.activealerts)
|
|
|
|
def add(self, alert_type, enabled=True, extra_text=''):
|
|
alert_type = str(alert_type)
|
|
this_alert = copy.copy(self.alerts[alert_type])
|
|
this_alert.alert_text_2 += extra_text
|
|
# downgrade the alert if we aren't enabled, except if it's FCW, which remains the same
|
|
# TODO: remove this 'if' by adding more alerts
|
|
if not enabled and this_alert.alert_type in [ET.WARNING, ET.SOFT_DISABLE, ET.IMMEDIATE_DISABLE] \
|
|
and this_alert != self.alerts['fcw']:
|
|
this_alert = alert("Comma Unavailable" if this_alert.alert_text_1 != "" else "", this_alert.alert_text_2, ET.NO_ENTRY, None, "chimeDouble", .4, 0., 3.)
|
|
|
|
# ignore no entries if we are enabled
|
|
if enabled and this_alert.alert_type in [ET.ENABLE, ET.NO_ENTRY]:
|
|
return
|
|
|
|
# if new alert is higher priority, log it
|
|
if self.current_alert is None or this_alert > self.current_alert:
|
|
cloudlog.event('alert_add',
|
|
alert_type=alert_type,
|
|
enabled=enabled)
|
|
|
|
self.activealerts.append(this_alert)
|
|
self.activealerts.sort()
|
|
|
|
def process_alerts(self, cur_time):
|
|
if self.alertPresent():
|
|
self.alert_start_time = cur_time
|
|
self.current_alert = self.activealerts[0]
|
|
print self.current_alert
|
|
alert_text_1 = ""
|
|
alert_text_2 = ""
|
|
visual_alert = "none"
|
|
audible_alert = "none"
|
|
|
|
if self.current_alert is not None:
|
|
# ewwwww
|
|
if self.alert_start_time + self.current_alert.duration_sound > cur_time:
|
|
audible_alert = self.current_alert.audible_alert
|
|
|
|
if self.alert_start_time + self.current_alert.duration_hud_alert > cur_time:
|
|
visual_alert = self.current_alert.visual_alert
|
|
|
|
if self.alert_start_time + self.current_alert.duration_text > cur_time:
|
|
alert_text_1 = self.current_alert.alert_text_1
|
|
alert_text_2 = self.current_alert.alert_text_2
|
|
|
|
# disable current alert
|
|
if self.alert_start_time + max(self.current_alert.duration_sound, self.current_alert.duration_hud_alert, self.current_alert.duration_text) < cur_time:
|
|
self.current_alert = None
|
|
|
|
# reset
|
|
self.activealerts = []
|
|
|
|
return alert_text_1, alert_text_2, visual_alert, audible_alert
|
|
|
|
|