brianhansonxyz/clearpilot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
Your Name
2024-04-27 13:44:34 -05:00
parent ea1aad5ed1
commit 2fbe9dbea1
25 changed files with 1501 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
selfdrive/controls/lib/events.py
View File

@@ -2,7 +2,7 @@
import math
import os
from enum import IntEnum
from typing import Dict, Union, Callable, List, Optional
from collections.abc import Callable
from cereal import log, car
import cereal.messaging as messaging
@@ -12,6 +12,7 @@ from openpilot.common.realtime import DT_CTRL
from openpilot.selfdrive.locationd.calibrationd import MIN_SPEED_FILTER
from openpilot.system.version import get_short_branch
params = Params()
params_memory = Params("/dev/shm/params")
AlertSize = log.ControlsState.AlertSize
@@ -51,12 +52,12 @@ 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.events: list[int] = []
self.static_events: list[int] = []
self.events_prev = dict.fromkeys(EVENTS.keys(), 0)
@property
def names(self) -> List[int]:
def names(self) -> list[int]:
return self.events
def __len__(self) -> int:
@@ -74,7 +75,7 @@ class Events:
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):
def create_alerts(self, event_types: list[str], callback_args=None):
if callback_args is None:
callback_args = []
@@ -135,7 +136,7 @@ class Alert:
self.creation_delay = creation_delay
self.alert_type = ""
self.event_type: Optional[str] = None
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}"
@@ -261,15 +262,14 @@ def no_gps_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, m
AlertStatus.normal, AlertSize.mid,
Priority.LOWER, VisualAlert.none, AudibleAlert.none, .2, creation_delay=300.)
def torque_nn_load_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int) -> Alert:
model_name = params_memory.get("NNFFModelName")
model_name = params.get("NNFFModelName", encoding='utf-8')
if model_name == "":
return Alert(
"NNFF Torque Controller not available",
"Donate logs to Twilsonco to get it added!",
"Donate logs to Twilsonco to get your car supported!",
AlertStatus.userPrompt, AlertSize.mid,
Priority.LOW, VisualAlert.none, AudibleAlert.prompt, 6.0)
Priority.LOW, VisualAlert.none, AudibleAlert.prompt, 5.0)
else:
return Alert(
"NNFF Torque Controller loaded",
@@ -379,13 +379,14 @@ def holiday_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster,
def no_lane_available_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int) -> Alert:
lane_width = sm['frogpilotPlan'].laneWidthLeft if CS.leftBlinker else sm['frogpilotPlan'].laneWidthRight
lane_width_msg = f"{lane_width:.1f} meters" if metric else f"{lane_width * CV.METER_TO_FOOT:.1f} feet"
return Alert(
"No lane available",
f"Detected lane width is only {lane_width_msg}",
AlertStatus.normal, AlertSize.mid,
Priority.LOWEST, VisualAlert.none, AudibleAlert.none, .2)
EVENTS: Dict[int, Dict[str, Union[Alert, AlertCallbackType]]] = {
EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
# ********** events with no alerts **********
EventName.stockFcw: {},
@@ -1009,7 +1010,13 @@ EVENTS: Dict[int, Dict[str, Union[Alert, AlertCallbackType]]] = {
},
# FrogPilot Events
EventName.frogSteerSaturated: {
EventName.blockUser: {
ET.PERMANENT: NormalPermanentAlert("Dashcam mode",
"Please don't use the 'Development' branch!",
priority=Priority.HIGHEST),
},
EventName.goatSteerSaturated: {
ET.WARNING: Alert(
"Turn Exceeds Steering Limit",
"JESUS TAKE THE WHEEL!!",
@@ -1110,6 +1117,14 @@ EVENTS: Dict[int, Dict[str, Union[Alert, AlertCallbackType]]] = {
Priority.LOW, VisualAlert.none, AudibleAlert.nessie, 4.),
},
EventName.accel40: {
ET.WARNING: Alert(
"Great Scott!",
"🚗💨",
AlertStatus.frogpilot, AlertSize.mid,
Priority.LOW, VisualAlert.none, AudibleAlert.doc, 4.),
},
EventName.firefoxSteerSaturated: {
ET.WARNING: Alert(
"Turn Exceeds Steering Limit",
@@ -1137,7 +1152,7 @@ EVENTS: Dict[int, Dict[str, Union[Alert, AlertCallbackType]]] = {
EventName.yourFrogTriedToKillMe: {
ET.PERMANENT: Alert(
"Your frog tried to kill me...",
"😡",
"👺",
AlertStatus.frogpilot, AlertSize.mid,
Priority.MID, VisualAlert.none, AudibleAlert.angry, 5.),
},
@@ -1150,7 +1165,7 @@ if __name__ == '__main__':
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))
alerts_by_type: dict[str, dict[Priority, list[str]]] = defaultdict(lambda: defaultdict(list))
CP = car.CarParams.new_message()
CS = car.CarState.new_message()
@@ -1162,7 +1177,7 @@ if __name__ == '__main__':
alert = alert(CP, CS, sm, False, 1)
alerts_by_type[et][alert.priority].append(event_names[i])
all_alerts: Dict[str, List[tuple[Priority, List[str]]]] = {}
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)