mirror of
https://github.com/sunnypilot/sunnypilot.git
synced 2026-09-30 21:23:43 +08:00
Revert "show MADS updates"
This reverts commitdaf0ad62Revert "fix changed events" This reverts commit31d8c97f
This commit is contained in:
DevTekVE
@@ -194,7 +194,6 @@ struct OnroadEvent @0xc4fa6047f024e718 {
|
||||
silentBrakeHold @157;
|
||||
silentWrongGear @158;
|
||||
silentReverseGear @159;
|
||||
madsStatusChanged @160;
|
||||
|
||||
soundsUnavailableDEPRECATED @47;
|
||||
}
|
||||
|
||||
@@ -222,40 +222,35 @@ def get_display_speed(speed_ms: float, metric: bool) -> str:
|
||||
|
||||
# ********** alert callback functions **********
|
||||
|
||||
AlertCallbackType = Callable[[car.CarParams, car.CarState, messaging.SubMaster, bool, int, log.ControlsState, tuple], Alert]
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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)}",
|
||||
"",
|
||||
@@ -263,8 +258,7 @@ def below_steer_speed_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.S
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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}%",
|
||||
@@ -275,43 +269,37 @@ def calibration_incomplete_alert(CP: car.CarParams, CS: car.CarState, sm: messag
|
||||
|
||||
# *** debug alerts ***
|
||||
|
||||
def out_of_space_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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)
|
||||
@@ -319,48 +307,41 @@ def calibration_invalid_alert(CP: car.CarParams, CS: car.CarState, sm: messaging
|
||||
return NormalPermanentAlert("Calibration Invalid", angles)
|
||||
|
||||
|
||||
def overheat_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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
|
||||
@@ -370,41 +351,11 @@ def longitudinal_maneuver_alert(CP: car.CarParams, CS: car.CarState, sm: messagi
|
||||
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,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
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)
|
||||
|
||||
|
||||
def mads_status_changed_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int, personality,
|
||||
mads_status: tuple[bool, bool]) -> Alert:
|
||||
lat_active, long_active = mads_status
|
||||
|
||||
lat_active_str = "ON" if lat_active else "OFF"
|
||||
long_active_str = "ON" if long_active else "OFF"
|
||||
|
||||
left_str = f"Steering: {lat_active_str}"
|
||||
right_str = f"ACC: {long_active_str}"
|
||||
|
||||
rec_width = 1848
|
||||
font_width_per_char = 20
|
||||
|
||||
left_str_width = len(left_str) * font_width_per_char
|
||||
right_str_width = len(right_str) * font_width_per_char
|
||||
pipe_width = font_width_per_char
|
||||
|
||||
total_padding = rec_width - (left_str_width + right_str_width + pipe_width)
|
||||
left_padding = total_padding // 2
|
||||
right_padding = total_padding - left_padding
|
||||
|
||||
final_str = (
|
||||
f"{' ' * (left_padding // font_width_per_char)}{left_str} "
|
||||
+ "| "
|
||||
+ f"{right_str}{' ' * (right_padding // font_width_per_char)}"
|
||||
)
|
||||
|
||||
return NormalPermanentAlert(final_str, duration=1.5)
|
||||
|
||||
|
||||
EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
|
||||
# ********** events with no alerts **********
|
||||
@@ -1094,10 +1045,6 @@ EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
|
||||
ET.NO_ENTRY: NoEntryAlert("Reverse Gear"),
|
||||
},
|
||||
|
||||
EventName.madsStatusChanged: {
|
||||
ET.WARNING: mads_status_changed_alert,
|
||||
},
|
||||
|
||||
}
|
||||
|
||||
|
||||
@@ -1116,7 +1063,7 @@ if __name__ == '__main__':
|
||||
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, (True, True))
|
||||
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]]]] = {}
|
||||
|
||||
@@ -424,8 +424,7 @@ class SelfdriveD:
|
||||
|
||||
pers = LONGITUDINAL_PERSONALITY_MAP[self.personality]
|
||||
alerts = self.events.create_alerts(self.state_machine.current_alert_types, [self.CP, CS, self.sm, self.is_metric,
|
||||
self.state_machine.soft_disable_timer, pers,
|
||||
(self.mads.enabled, self.enabled)])
|
||||
self.state_machine.soft_disable_timer, pers])
|
||||
self.AM.add_many(self.sm.frame, alerts)
|
||||
self.AM.process_alerts(self.sm.frame, clear_event_types)
|
||||
|
||||
|
||||
+11
-13
@@ -17,7 +17,6 @@ class ModularAssistiveDrivingSystem:
|
||||
|
||||
self.enabled = False
|
||||
self.active = False
|
||||
self.enabled_prev = False
|
||||
self.available = False
|
||||
self.allow_always = False
|
||||
self.selfdrive = selfdrive
|
||||
@@ -36,12 +35,12 @@ class ModularAssistiveDrivingSystem:
|
||||
|
||||
def update_events(self, CS: car.CarState):
|
||||
def update_unified_engagement_mode():
|
||||
if (self.unified_engagement_mode and self.enabled) or not self.unified_engagement_mode:
|
||||
if (self.unified_engagement_mode and self.active) or not self.unified_engagement_mode:
|
||||
self.selfdrive.events.remove(EventName.pcmEnable)
|
||||
self.selfdrive.events.remove(EventName.buttonEnable)
|
||||
|
||||
def update_silent_lkas_enable():
|
||||
if self.state_machine.state == State.paused and self.enabled:
|
||||
if self.state_machine.state == State.paused and self.active:
|
||||
self.selfdrive.events.add(EventName.silentLkasEnable)
|
||||
|
||||
if not self.selfdrive.enabled:
|
||||
@@ -53,8 +52,6 @@ class ModularAssistiveDrivingSystem:
|
||||
if not self.selfdrive.events.has(EventName.silentWrongGear) and not self.selfdrive.events.has(EventName.silentReverseGear):
|
||||
update_silent_lkas_enable()
|
||||
|
||||
self.selfdrive.events.remove(EventName.preEnableStandstill)
|
||||
|
||||
if self.disengage_lateral_on_brake_toggle:
|
||||
if self.selfdrive.events.has(EventName.brakeHold):
|
||||
self.selfdrive.events.replace(EventName.brakeHold, EventName.silentBrakeHold)
|
||||
@@ -73,9 +70,15 @@ class ModularAssistiveDrivingSystem:
|
||||
self.selfdrive.events.add(EventName.lkasEnable)
|
||||
|
||||
for be in CS.buttonEvents:
|
||||
if be.type == ButtonType.cancel:
|
||||
if not self.selfdrive.enabled and self.selfdrive.enabled_prev:
|
||||
self.selfdrive.events.add(EventName.manualLongitudinalRequired)
|
||||
if be.type == ButtonType.lkas and be.pressed and (CS.cruiseState.available or self.allow_always):
|
||||
if self.enabled:
|
||||
self.selfdrive.events.add(EventName.lkasDisable)
|
||||
if self.active:
|
||||
if self.selfdrive.enabled:
|
||||
self.selfdrive.events.add(EventName.manualSteeringRequired)
|
||||
else:
|
||||
self.selfdrive.events.add(EventName.lkasDisable)
|
||||
else:
|
||||
self.selfdrive.events.add(EventName.lkasEnable)
|
||||
|
||||
@@ -99,10 +102,5 @@ class ModularAssistiveDrivingSystem:
|
||||
if not self.selfdrive.CP.passive and self.selfdrive.initialized:
|
||||
self.enabled, self.active = self.state_machine.update(self.selfdrive.events)
|
||||
|
||||
# update MADS status events
|
||||
if self.selfdrive.enabled != self.selfdrive.enabled_prev or self.enabled != self.enabled_prev:
|
||||
self.selfdrive.events.add(EventName.madsStatusChanged)
|
||||
|
||||
# Copy of previous SelfdriveD and MADS states for MADS events handling
|
||||
# Copy of previous SelfdriveD states for MADS events handling
|
||||
self.selfdrive.enabled_prev = self.selfdrive.enabled
|
||||
self.enabled_prev = self.enabled
|
||||
|
||||
Reference in New Issue
Block a user