system/ui: add centralized UIState singleton for global state management (#35397)

* add centralized UIState singleton for global state management

* safely import ui_state

* merge master

* merge master

system/ui/lib/application.py

@@ -36,6 +36,16 @@ class FontWeight(IntEnum):
   BLACK = 8
 
 
+def _get_ui_state():
+  """Safely import and return ui_state, or None if unavailable."""
+  try:
+    from openpilot.system.ui.lib.ui_state import ui_state
+
+    return ui_state
+  except ImportError:
+    return None
+
+
 class GuiApplication:
   def __init__(self, width: int, height: int):
     self._fonts: dict[FontWeight, rl.Font] = {}
@@ -139,6 +149,7 @@ class GuiApplication:
     rl.close_window()
 
   def render(self):
+    ui_state = _get_ui_state()
     try:
       while not (self._window_close_requested or rl.window_should_close()):
         if self._render_texture:
@@ -148,6 +159,9 @@ class GuiApplication:
           rl.begin_drawing()
           rl.clear_background(rl.BLACK)
 
+        if ui_state is not None:
+          ui_state.update()
+
         yield
 
         if self._render_texture:

system/ui/lib/ui_state.py

@@ -0,0 +1,129 @@
+import pyray as rl
+from enum import Enum
+from cereal import messaging, log
+from openpilot.common.params import Params, UnknownKeyName
+
+
+UI_BORDER_SIZE = 30
+
+
+class UIStatus(Enum):
+  DISENGAGED = "disengaged"
+  ENGAGED = "engaged"
+  OVERRIDE = "override"
+
+
+class UIState:
+  _instance: 'UIState | None' = None
+
+  def __new__(cls):
+    if cls._instance is None:
+      cls._instance = super().__new__(cls)
+      cls._instance._initialize()
+    return cls._instance
+
+  def _initialize(self):
+    self.params = Params()
+    self.sm = messaging.SubMaster(
+      [
+        "modelV2",
+        "controlsState",
+        "liveCalibration",
+        "radarState",
+        "deviceState",
+        "pandaStates",
+        "carParams",
+        "driverMonitoringState",
+        "carState",
+        "driverStateV2",
+        "roadCameraState",
+        "wideRoadCameraState",
+        "managerState",
+        "selfdriveState",
+        "longitudinalPlan",
+      ]
+    )
+
+    # UI Status tracking
+    self.status: UIStatus = UIStatus.DISENGAGED
+    self.started_frame: int = 0
+    self._engaged_prev: bool = False
+    self._started_prev: bool = False
+
+    # Core state variables
+    self.is_metric: bool = self.params.get_bool("IsMetric")
+    self.started: bool = False
+    self.ignition: bool = False
+    self.panda_type: log.PandaState.PandaType = log.PandaState.PandaType.unknown
+    self.personality: log.LongitudinalPersonality = log.LongitudinalPersonality.standard
+    self.light_sensor: float = -1.0
+
+    self._update_params()
+
+  @property
+  def engaged(self) -> bool:
+    return self.started and self.sm["selfdriveState"].enabled
+
+  def update(self) -> None:
+    self.sm.update(0)
+    self._update_state()
+    self._update_status()
+
+  def _update_state(self) -> None:
+    # Handle panda states updates
+    if self.sm.updated["pandaStates"]:
+      panda_states = self.sm["pandaStates"]
+
+      if len(panda_states) > 0:
+        # Get panda type from first panda
+        self.panda_type = panda_states[0].pandaType
+        # Check ignition status across all pandas
+        if self.panda_type != log.PandaState.PandaType.unknown:
+          self.ignition = any(state.ignitionLine or state.ignitionCan for state in panda_states)
+    elif self.sm.frame - self.sm.recv_frame["pandaStates"] > 5 * rl.get_fps():
+      self.panda_type = log.PandaState.PandaType.unknown
+
+    # Handle wide road camera state updates
+    if self.sm.updated["wideRoadCameraState"]:
+      cam_state = self.sm["wideRoadCameraState"]
+
+      # Scale factor based on sensor type
+      scale = 6.0 if cam_state.sensor == 'ar0231' else 1.0
+      self.light_sensor = max(100.0 - scale * cam_state.exposureValPercent, 0.0)
+    elif not self.sm.alive["wideRoadCameraState"] or not self.sm.valid["wideRoadCameraState"]:
+      self.light_sensor = -1
+
+    # Update started state
+    self.started = self.sm["deviceState"].started and self.ignition
+
+  def _update_status(self) -> None:
+    if self.started and self.sm.updated["selfdriveState"]:
+      ss = self.sm["selfdriveState"]
+      state = ss.state
+
+      if state in (log.SelfdriveState.OpenpilotState.preEnabled, log.SelfdriveState.OpenpilotState.overriding):
+        self.status = UIStatus.OVERRIDE
+      else:
+        self.status = UIStatus.ENGAGED if ss.enabled else UIStatus.DISENGAGED
+
+    # Check for engagement state changes
+    if self.engaged != self._engaged_prev:
+      self._engaged_prev = self.engaged
+
+    # Handle onroad/offroad transition
+    if self.started != self._started_prev or self.sm.frame == 1:
+      if self.started:
+        self.status = UIStatus.DISENGAGED
+        self.started_frame = self.sm.frame
+
+      self._started_prev = self.started
+
+  def _update_params(self) -> None:
+    try:
+      self.is_metric = self.params.get_bool("IsMetric")
+    except UnknownKeyName:
+      self.is_metric = False
+
+
+# Global instance
+ui_state = UIState()

system/ui/onroad/alert_renderer.py

@@ -5,6 +5,7 @@ from cereal import messaging, log
 from openpilot.system.hardware import TICI
 from openpilot.system.ui.lib.application import gui_app, FontWeight
 from openpilot.system.ui.lib.label import gui_text_box
+from openpilot.system.ui.lib.ui_state import ui_state
 
 
 ALERT_MARGIN = 40
@@ -61,8 +62,6 @@ ALERT_CRITICAL_REBOOT = Alert(
 
 class AlertRenderer:
   def __init__(self):
-    # TODO: use ui_state to determine when to start
-    self.started_frame: int = 0
     self.font_regular: rl.Font = gui_app.font(FontWeight.NORMAL)
     self.font_bold: rl.Font = gui_app.font(FontWeight.BOLD)
     self.font_metrics_cache: dict[tuple[str, int, str], rl.Vector2] = {}
@@ -72,7 +71,7 @@ class AlertRenderer:
     ss = sm['selfdriveState']
 
     # Check if waiting to start
-    if sm.recv_frame['selfdriveState'] < self.started_frame:
+    if sm.recv_frame['selfdriveState'] < ui_state.started_frame:
       return ALERT_STARTUP_PENDING
 
     # Handle selfdrive timeout

system/ui/onroad/augmented_road_view.py

@@ -1,9 +1,9 @@
 import numpy as np
 import pyray as rl
-from enum import Enum
 
-from cereal import messaging, log
+from cereal import log
 from msgq.visionipc import VisionStreamType
+from openpilot.system.ui.lib.ui_state import ui_state, UIStatus, UI_BORDER_SIZE
 from openpilot.system.ui.onroad.alert_renderer import AlertRenderer
 from openpilot.system.ui.onroad.driver_state import DriverStateRenderer
 from openpilot.system.ui.onroad.hud_renderer import HudRenderer
@@ -17,19 +17,18 @@ from openpilot.common.transformations.orientation import rot_from_euler
 OpState = log.SelfdriveState.OpenpilotState
 CALIBRATED = log.LiveCalibrationData.Status.calibrated
 DEFAULT_DEVICE_CAMERA = DEVICE_CAMERAS["tici", "ar0231"]
-UI_BORDER_SIZE = 30
 
-class BorderStatus(Enum):
-  DISENGAGED = rl.Color(0x17, 0x33, 0x49, 0xc8) # Blue for disengaged state
-  OVERRIDE = rl.Color(0x91, 0x9b, 0x95, 0xf1)   # Gray for override state
-  ENGAGED = rl.Color(0x17, 0x86, 0x44, 0xf1)    # Green for engaged state
+BORDER_COLORS = {
+  UIStatus.DISENGAGED: rl.Color(0x17, 0x33, 0x49, 0xC8),   # Blue for disengaged state
+  UIStatus.OVERRIDE: rl.Color(0x91, 0x9B, 0x95, 0xF1),     # Gray for override state
+  UIStatus.ENGAGED: rl.Color(0x17, 0x86, 0x44, 0xF1),      # Green for engaged state
+}
 
 
 class AugmentedRoadView(CameraView):
-  def __init__(self, sm: messaging.SubMaster, stream_type: VisionStreamType):
+  def __init__(self, stream_type: VisionStreamType):
     super().__init__("camerad", stream_type)
 
-    self.sm = sm
     self.device_camera: DeviceCameraConfig | None = None
     self.view_from_calib = view_frame_from_device_frame.copy()
     self.view_from_wide_calib = view_frame_from_device_frame.copy()
@@ -72,10 +71,10 @@ class AugmentedRoadView(CameraView):
     super().render(rect)
 
     # Draw all UI overlays
-    self.model_renderer.draw(self._content_rect, self.sm)
-    self._hud_renderer.draw(self._content_rect, self.sm)
-    self.alert_renderer.draw(self._content_rect, self.sm)
-    self.driver_state_renderer.draw(self._content_rect, self.sm)
+    self.model_renderer.draw(self._content_rect, ui_state.sm)
+    self._hud_renderer.draw(self._content_rect, ui_state.sm)
+    self.alert_renderer.draw(self._content_rect, ui_state.sm)
+    self.driver_state_renderer.draw(self._content_rect, ui_state.sm)
 
     # Custom UI extension point - add custom overlays here
     # Use self._content_rect for positioning within camera bounds
@@ -84,18 +83,12 @@ class AugmentedRoadView(CameraView):
     rl.end_scissor_mode()
 
   def _draw_border(self, rect: rl.Rectangle):
-    state = self.sm["selfdriveState"]
-    if state.state in (OpState.preEnabled, OpState.overriding):
-      status = BorderStatus.OVERRIDE
-    elif state.enabled:
-      status = BorderStatus.ENGAGED
-    else:
-      status = BorderStatus.DISENGAGED
-
-    rl.draw_rectangle_lines_ex(rect, UI_BORDER_SIZE, status.value)
+    border_color = BORDER_COLORS.get(ui_state.status, BORDER_COLORS[UIStatus.DISENGAGED])
+    rl.draw_rectangle_lines_ex(rect, UI_BORDER_SIZE, border_color)
 
   def _update_calibration(self):
     # Update device camera if not already set
+    sm = ui_state.sm
     if not self.device_camera and sm.seen['roadCameraState'] and sm.seen['deviceState']:
       self.device_camera = DEVICE_CAMERAS[(str(sm['deviceState'].deviceType), str(sm['roadCameraState'].sensor))]
 
@@ -103,7 +96,7 @@ class AugmentedRoadView(CameraView):
     if not (sm.updated["liveCalibration"] and sm.valid['liveCalibration']):
       return
 
-    calib = self.sm['liveCalibration']
+    calib = sm['liveCalibration']
     if len(calib.rpyCalib) != 3 or calib.calStatus != CALIBRATED:
       return
 
@@ -118,7 +111,7 @@ class AugmentedRoadView(CameraView):
 
   def _calc_frame_matrix(self, rect: rl.Rectangle) -> np.ndarray:
     # Check if we can use cached matrix
-    calib_time = self.sm.recv_frame['liveCalibration']
+    calib_time = ui_state.sm.recv_frame['liveCalibration']
     current_dims = (self._content_rect.width, self._content_rect.height)
     if (self._last_calib_time == calib_time and
         self._last_rect_dims == current_dims and
@@ -178,14 +171,10 @@ class AugmentedRoadView(CameraView):
 
 if __name__ == "__main__":
   gui_app.init_window("OnRoad Camera View")
-  sm = messaging.SubMaster(["modelV2", "controlsState", "liveCalibration", "radarState", "deviceState",
-    "pandaStates", "carParams", "driverMonitoringState", "carState", "driverStateV2",
-    "roadCameraState", "wideRoadCameraState", "managerState", "selfdriveState", "longitudinalPlan"])
-  road_camera_view = AugmentedRoadView(sm, VisionStreamType.VISION_STREAM_ROAD)
+  road_camera_view = AugmentedRoadView(VisionStreamType.VISION_STREAM_ROAD)
   print("***press space to switch camera view***")
   try:
     for _ in gui_app.render():
-      sm.update(0)
       if rl.is_key_released(rl.KeyboardKey.KEY_SPACE):
         is_wide = road_camera_view.stream_type == VisionStreamType.VISION_STREAM_WIDE_ROAD
         road_camera_view.switch_stream(VisionStreamType.VISION_STREAM_ROAD if is_wide else VisionStreamType.VISION_STREAM_WIDE_ROAD)

system/ui/onroad/driver_state.py

@@ -2,6 +2,8 @@ import numpy as np
 import pyray as rl
 from dataclasses import dataclass
 from openpilot.system.ui.lib.application import gui_app
+from openpilot.system.ui.lib.ui_state import ui_state, UI_BORDER_SIZE
+
 
 # Default 3D coordinates for face keypoints as a NumPy array
 DEFAULT_FACE_KPTS_3D = np.array([
@@ -17,7 +19,6 @@ DEFAULT_FACE_KPTS_3D = np.array([
 ], dtype=np.float32)
 
 # UI constants
-UI_BORDER_SIZE = 30
 BTN_SIZE = 192
 IMG_SIZE = 144
 ARC_LENGTH = 133
@@ -95,8 +96,7 @@ class DriverStateRenderer:
     rl.draw_spline_linear(self.face_lines, len(self.face_lines), 5.2, self.white_color)
 
     # Set arc color based on engaged state
-    engaged = True
-    self.arc_color = self.engaged_color if engaged else self.disengaged_color
+    self.arc_color = self.engaged_color if ui_state.engaged else self.disengaged_color
     self.arc_color.a = int(0.4 * 255 * (1.0 - self.dm_fade_state))  # Fade out when inactive
 
     # Draw arcs
@@ -107,7 +107,7 @@ class DriverStateRenderer:
 
   def _is_visible(self, sm):
     """Check if the visualization should be rendered."""
-    return (sm.seen['driverStateV2'] and
+    return (sm.recv_frame['driverStateV2'] > ui_state.started_frame and
             sm.seen['driverMonitoringState'] and
             sm['selfdriveState'].alertSize == 0)
 

system/ui/onroad/hud_renderer.py

@@ -1,9 +1,9 @@
 import pyray as rl
 from dataclasses import dataclass
 from cereal.messaging import SubMaster
+from openpilot.system.ui.lib.ui_state import ui_state, UIStatus
 from openpilot.system.ui.lib.application import gui_app, FontWeight
 from openpilot.common.conversions import Conversions as CV
-from enum import IntEnum
 
 # Constants
 SET_SPEED_NA = 255
@@ -53,17 +53,9 @@ FONT_SIZES = FontSizes()
 COLORS = Colors()
 
 
-class HudStatus(IntEnum):
-  DISENGAGED = 0
-  OVERRIDE = 1
-  ENGAGED = 2
-
-
 class HudRenderer:
   def __init__(self):
     """Initialize the HUD renderer."""
-    self.is_metric: bool = False
-    self.status: HudStatus = HudStatus.DISENGAGED
     self.is_cruise_set: bool = False
     self.is_cruise_available: bool = False
     self.set_speed: float = SET_SPEED_NA
@@ -77,10 +69,7 @@ class HudRenderer:
 
   def _update_state(self, sm: SubMaster) -> None:
     """Update HUD state based on car state and controls state."""
-    self.is_metric = True
-    self.status = HudStatus.DISENGAGED
-
-    if not sm.valid['carState']:
+    if sm.recv_frame["carState"] < ui_state.started_frame:
       self.is_cruise_set = False
       self.set_speed = SET_SPEED_NA
       self.speed = 0.0
@@ -96,13 +85,13 @@ class HudRenderer:
     self.is_cruise_set = 0 < self.set_speed < SET_SPEED_NA
     self.is_cruise_available = self.set_speed != -1
 
-    if self.is_cruise_set and not self.is_metric:
+    if self.is_cruise_set and not ui_state.is_metric:
       self.set_speed *= KM_TO_MILE
 
     v_ego_cluster = car_state.vEgoCluster
     self.v_ego_cluster_seen = self.v_ego_cluster_seen or v_ego_cluster != 0.0
     v_ego = v_ego_cluster if self.v_ego_cluster_seen else car_state.vEgo
-    speed_conversion = CV.MS_TO_KPH if self.is_metric else CV.MS_TO_MPH
+    speed_conversion = CV.MS_TO_KPH if ui_state.is_metric else CV.MS_TO_MPH
     self.speed = max(0.0, v_ego * speed_conversion)
 
   def draw(self, rect: rl.Rectangle, sm: SubMaster) -> None:
@@ -125,7 +114,7 @@ class HudRenderer:
 
   def _draw_set_speed(self, rect: rl.Rectangle) -> None:
     """Draw the MAX speed indicator box."""
-    set_speed_width = UI_CONFIG.set_speed_width_metric if self.is_metric else UI_CONFIG.set_speed_width_imperial
+    set_speed_width = UI_CONFIG.set_speed_width_metric if ui_state.is_metric else UI_CONFIG.set_speed_width_imperial
     x = rect.x + 60 + (UI_CONFIG.set_speed_width_imperial - set_speed_width) // 2
     y = rect.y + 45
 
@@ -137,11 +126,12 @@ class HudRenderer:
     set_speed_color = COLORS.dark_grey
     if self.is_cruise_set:
       set_speed_color = COLORS.white
-      max_color = {
-        HudStatus.DISENGAGED: COLORS.disengaged,
-        HudStatus.OVERRIDE: COLORS.override,
-        HudStatus.ENGAGED: COLORS.engaged,
-      }.get(self.status, COLORS.grey)
+      if ui_state.status == UIStatus.ENGAGED:
+        max_color = COLORS.engaged
+      elif ui_state.status == UIStatus.DISENGAGED:
+        max_color = COLORS.disengaged
+      elif ui_state.status == UIStatus.OVERRIDE:
+        max_color = COLORS.override
 
     max_text = "MAX"
     max_text_width = self._measure_text(max_text, self._font_semi_bold, FONT_SIZES.max_speed, 'semi_bold').x
@@ -172,7 +162,7 @@ class HudRenderer:
     speed_pos = rl.Vector2(rect.x + rect.width / 2 - speed_text_size.x / 2, 180 - speed_text_size.y / 2)
     rl.draw_text_ex(self._font_bold, speed_text, speed_pos, FONT_SIZES.current_speed, 0, COLORS.white)
 
-    unit_text = "km/h" if self.is_metric else "mph"
+    unit_text = "km/h" if ui_state.is_metric else "mph"
     unit_text_size = self._measure_text(unit_text, self._font_medium, FONT_SIZES.speed_unit, 'medium')
     unit_pos = rl.Vector2(rect.x + rect.width / 2 - unit_text_size.x / 2, 290 - unit_text_size.y / 2)
     rl.draw_text_ex(self._font_medium, unit_text, unit_pos, FONT_SIZES.speed_unit, 0, COLORS.white_translucent)
@@ -183,7 +173,7 @@ class HudRenderer:
     center_y = int(rect.y + UI_CONFIG.border_size + UI_CONFIG.button_size / 2)
     rl.draw_circle(center_x, center_y, UI_CONFIG.button_size / 2, COLORS.black_translucent)
 
-    opacity = 0.7 if self.status == HudStatus.DISENGAGED else 1.0
+    opacity = 0.7 if ui_state.status == UIStatus.DISENGAGED else 1.0
     img_pos = rl.Vector2(center_x - self._wheel_texture.width / 2, center_y - self._wheel_texture.height / 2)
     rl.draw_texture_v(self._wheel_texture, img_pos, rl.Color(255, 255, 255, int(255 * opacity)))
 

system/ui/onroad/model_renderer.py

@@ -5,6 +5,7 @@ from cereal import messaging, car
 from dataclasses import dataclass, field
 from openpilot.common.params import Params
 from openpilot.system.ui.lib.application import DEFAULT_FPS
+from openpilot.system.ui.lib.ui_state import ui_state
 from openpilot.system.ui.lib.shader_polygon import draw_polygon
 
 
@@ -86,7 +87,8 @@ class ModelRenderer:
 
   def draw(self, rect: rl.Rectangle, sm: messaging.SubMaster):
     # Check if data is up-to-date
-    if not sm.valid['modelV2'] or not sm.valid['liveCalibration']:
+    if (sm.recv_frame["liveCalibration"] < ui_state.started_frame or
+        sm.recv_frame["modelV2"] < ui_state.started_frame):
       return
 
     # Set up clipping region