refactor AlertManager

selfdrive/controls/controlsd.py

@@ -619,8 +619,9 @@ def publish_logs(self, CS, start_time, actuators, lac_log):
     clear_event = ET.WARNING if ET.WARNING not in self.current_alert_types else None
     alerts = self.events.create_alerts(self.current_alert_types, [self.CP, self.sm, self.is_metric, self.soft_disable_timer])
     self.AM.add_many(self.sm.frame, alerts)
-    self.AM.process_alerts(self.sm.frame, clear_event)
-    hudControl.visualAlert = self.AM.visual_alert
+    current_alert = self.AM.process_alerts(self.sm.frame, clear_event)
+    if current_alert:
+      hudControl.visualAlert = current_alert.visual_alert
 
     if not self.read_only and self.initialized:
       # send car controls over can
@@ -641,13 +642,15 @@ def publish_logs(self, CS, start_time, actuators, lac_log):
     dat = messaging.new_message('controlsState')
     dat.valid = CS.canValid
     controlsState = dat.controlsState
-    controlsState.alertText1 = self.AM.alert_text_1
-    controlsState.alertText2 = self.AM.alert_text_2
-    controlsState.alertSize = self.AM.alert_size
-    controlsState.alertStatus = self.AM.alert_status
-    controlsState.alertBlinkingRate = self.AM.alert_rate
-    controlsState.alertType = self.AM.alert_type
-    controlsState.alertSound = self.AM.audible_alert
+    if current_alert:
+      controlsState.alertText1 = current_alert.alert_text_1
+      controlsState.alertText2 = current_alert.alert_text_2
+      controlsState.alertSize = current_alert.alert_size
+      controlsState.alertStatus = current_alert.alert_status
+      controlsState.alertBlinkingRate = current_alert.alert_rate
+      controlsState.alertType = current_alert.alert_type
+      controlsState.alertSound = current_alert.audible_alert
+
     controlsState.canMonoTimes = list(CS.canMonoTimes)
     controlsState.longitudinalPlanMonoTime = self.sm.logMonoTime['longitudinalPlan']
     controlsState.lateralPlanMonoTime = self.sm.logMonoTime['lateralPlan']

selfdrive/controls/lib/alertmanager.py

@@ -5,7 +5,6 @@
 from dataclasses import dataclass
 from typing import List, Dict, Optional
 
-from cereal import car, log
 from common.basedir import BASEDIR
 from common.params import Params
 from selfdrive.controls.lib.events import Alert
@@ -36,22 +35,9 @@ def active(self, frame: int) -> bool:
     return frame <= self.end_frame
 
 class AlertManager:
-
   def __init__(self):
-    self.reset()
     self.alerts: Dict[str, AlertEntry] = defaultdict(AlertEntry)
 
-  def reset(self) -> None:
-    self.alert: Optional[Alert] = None
-    self.alert_type: str = ""
-    self.alert_text_1: str = ""
-    self.alert_text_2: str = ""
-    self.alert_status = log.ControlsState.AlertStatus.normal
-    self.alert_size = log.ControlsState.AlertSize.none
-    self.visual_alert = car.CarControl.HUDControl.VisualAlert.none
-    self.audible_alert = car.CarControl.HUDControl.AudibleAlert.none
-    self.alert_rate: float = 0.
-
   def add_many(self, frame: int, alerts: List[Alert]) -> None:
     for alert in alerts:
       key = alert.alert_type
@@ -61,30 +47,18 @@ def add_many(self, frame: int, alerts: List[Alert]) -> None:
       min_end_frame = self.alerts[key].start_frame + alert.duration
       self.alerts[key].end_frame = max(frame + 1, min_end_frame)
 
-  def process_alerts(self, frame: int, clear_event_type=None) -> None:
+  def process_alerts(self, frame: int, clear_event_type=None) -> Optional[Alert]:
     current_alert = AlertEntry()
-    for k, v in self.alerts.items():
-      if v.alert is None:
+    for v in self.alerts.values():
+      if not v.alert:
         continue
 
-      if clear_event_type is not None and v.alert.event_type == clear_event_type:
-        self.alerts[k].end_frame = -1
+      if clear_event_type and v.alert.event_type == clear_event_type:
+        v.end_frame = -1
 
       # sort by priority first and then by start_frame
       greater = current_alert.alert is None or (v.alert.priority, v.start_frame) > (current_alert.alert.priority, current_alert.start_frame)
       if v.active(frame) and greater:
         current_alert = v
 
-    # clear current alert
-    self.reset()
-
-    self.alert = current_alert.alert
-    if self.alert is not None:
-      self.alert_type = self.alert.alert_type
-      self.audible_alert = self.alert.audible_alert
-      self.visual_alert = self.alert.visual_alert
-      self.alert_text_1 = self.alert.alert_text_1
-      self.alert_text_2 = self.alert.alert_text_2
-      self.alert_status = self.alert.alert_status
-      self.alert_size = self.alert.alert_size
-      self.alert_rate = self.alert.alert_rate
+    return current_alert.alert

selfdrive/controls/lib/tests/test_alertmanager.py

@@ -34,9 +34,9 @@ def test_duration(self):
         for frame in range(duration+10):
           if frame < add_duration:
             AM.add_many(frame, [alert, ])
-          AM.process_alerts(frame)
+          current_alert = AM.process_alerts(frame)
 
-          shown = AM.alert is not None
+          shown = current_alert is not None
           should_show = frame <= show_duration
           self.assertEqual(shown, should_show, msg=f"{frame=} {add_duration=} {duration=}")