ui: add SingleCallback for safe, flexible single-callback management in UI components

selfdrive/ui/mici/layouts/onboarding.py

@@ -102,10 +102,10 @@ def wrapped_continue_callback():
     self._dialog = DriverCameraSetupDialog(wrapped_continue_callback)
 
     # Disable driver monitoring model when device times out for inactivity
-    def inactivity_callback():
-      ui_state.params.put_bool("IsDriverViewEnabled", False)
+    device.add_interactive_timeout_callback(self.inactivity_callback)
 
-    device.add_interactive_timeout_callback(inactivity_callback)
+  def inactivity_callback(self):
+    ui_state.params.put_bool("IsDriverViewEnabled", False)
 
   def show_event(self):
     super().show_event()
@@ -219,7 +219,7 @@ def on_continue():
     self._steps = [
       TrainingGuideAttentionNotice(continue_callback=on_continue),
       TrainingGuidePreDMTutorial(continue_callback=on_continue),
-      TrainingGuideDMTutorial(continue_callback=on_continue),
+      # TrainingGuideDMTutorial(continue_callback=on_continue),
       TrainingGuideRecordFront(continue_callback=on_continue),
     ]
 

selfdrive/ui/mici/onroad/driver_camera_dialog.py

@@ -34,7 +34,7 @@ def __init__(self, no_escape=False):
     self._pm: messaging.PubMaster | None = None
     if not no_escape:
       # TODO: this can grow unbounded, should be given some thought
-      device.add_interactive_timeout_callback(lambda: gui_app.set_modal_overlay(None))
+      device.add_interactive_timeout_callback(self.close_dialog)
     self.set_back_callback(lambda: gui_app.set_modal_overlay(None))
     self.set_back_enabled(not no_escape)
 
@@ -60,6 +60,9 @@ def hide_event(self):
     ui_state.params.put_bool("IsDriverViewEnabled", False)
     device.reset_interactive_timeout()
 
+  def close_dialog(self):
+    gui_app.set_modal_overlay(None)
+
   def _handle_mouse_release(self, _):
     ui_state.params.remove("DriverTooDistracted")
 

selfdrive/ui/mici/widgets/dialog.py

@@ -8,6 +8,7 @@
 from openpilot.system.ui.widgets import Widget, NavWidget, DialogResult
 from openpilot.system.ui.widgets.label import UnifiedLabel, gui_label
 from openpilot.system.ui.widgets.mici_keyboard import MiciKeyboard
+from openpilot.system.ui.lib.callback_manager import SingleCallback
 from openpilot.system.ui.lib.text_measure import measure_text_cached
 from openpilot.system.ui.lib.wrap_text import wrap_text
 from openpilot.system.ui.lib.application import gui_app, FontWeight, MousePos
@@ -16,6 +17,7 @@
 from openpilot.common.filter_simple import FirstOrderFilter
 from openpilot.selfdrive.ui.mici.widgets.button import BigButton
 
+
 DEBUG = False
 
 PADDING = 20
@@ -26,7 +28,7 @@ def __init__(self, right_btn: str | None = None, right_btn_callback: Callable |
     super().__init__()
     self._ret = DialogResult.NO_ACTION
     self.set_rect(rl.Rectangle(0, 0, gui_app.width, gui_app.height))
-    self.set_back_callback(lambda: setattr(self, '_ret', DialogResult.CANCEL))
+    self.set_back_callback(self._on_back_pressed)
 
     self._right_btn = None
     if right_btn:
@@ -40,6 +42,9 @@ def right_btn_callback_wrapper():
       # move to right side
       self._right_btn._rect.x = self._rect.x + self._rect.width - self._right_btn._rect.width
 
+  def _on_back_pressed(self):
+    self._ret = DialogResult.CANCEL
+
   def _render(self, _) -> DialogResult:
     """
     Allows `gui_app.set_modal_overlay(BigDialog(...))`.
@@ -102,7 +107,7 @@ def __init__(self, title: str, icon: str, red: bool = False,
                exit_on_confirm: bool = True,
                confirm_callback: Callable | None = None):
     super().__init__()
-    self._confirm_callback = confirm_callback
+    self._confirm_callback = SingleCallback(confirm_callback)
     self._exit_on_confirm = exit_on_confirm
 
     icon_txt = gui_app.texture(icon, 64, 53)

selfdrive/ui/ui_state.py

@@ -10,6 +10,7 @@
 from openpilot.common.swaglog import cloudlog
 from openpilot.selfdrive.ui.lib.prime_state import PrimeState
 from openpilot.system.ui.lib.application import gui_app
+from openpilot.system.ui.lib.callback_manager import CallbackManager
 from openpilot.system.hardware import HARDWARE, PC
 
 BACKLIGHT_OFFROAD = 65 if HARDWARE.get_device_type() == "mici" else 50
@@ -82,16 +83,16 @@ def _initialize(self):
     self._param_update_time: float = 0.0
 
     # Callbacks
-    self._offroad_transition_callbacks: list[Callable[[], None]] = []
-    self._engaged_transition_callbacks: list[Callable[[], None]] = []
+    self._offroad_transition_callbacks = CallbackManager()
+    self._engaged_transition_callbacks = CallbackManager()
 
     self.update_params()
 
   def add_offroad_transition_callback(self, callback: Callable[[], None]):
-    self._offroad_transition_callbacks.append(callback)
+    self._offroad_transition_callbacks.add(callback)
 
   def add_engaged_transition_callback(self, callback: Callable[[], None]):
-    self._engaged_transition_callbacks.append(callback)
+    self._engaged_transition_callbacks.add(callback)
 
   @property
   def engaged(self) -> bool:
@@ -154,8 +155,7 @@ def _update_status(self) -> None:
 
     # Check for engagement state changes
     if self.engaged != self._engaged_prev:
-      for callback in self._engaged_transition_callbacks:
-        callback()
+      self._offroad_transition_callbacks()
       self._engaged_prev = self.engaged
 
     # Handle onroad/offroad transition
@@ -165,8 +165,7 @@ def _update_status(self) -> None:
         self.started_frame = self.sm.frame
         self.started_time = time.monotonic()
 
-      for callback in self._offroad_transition_callbacks:
-        callback()
+      self._offroad_transition_callbacks()
 
       self._started_prev = self.started
 
@@ -187,7 +186,7 @@ class Device:
   def __init__(self):
     self._ignition = False
     self._interaction_time: float = -1
-    self._interactive_timeout_callbacks: list[Callable] = []
+    self._interactive_timeout_callbacks = CallbackManager()
     self._prev_timed_out = False
     self._awake: bool = True
 
@@ -207,7 +206,7 @@ def reset_interactive_timeout(self, timeout: int = -1) -> None:
     self._interaction_time = time.monotonic() + timeout
 
   def add_interactive_timeout_callback(self, callback: Callable):
-    self._interactive_timeout_callbacks.append(callback)
+    self._interactive_timeout_callbacks.add(callback)
 
   def update(self):
     # do initial reset
@@ -256,8 +255,7 @@ def _update_wakefulness(self):
 
     interaction_timeout = time.monotonic() > self._interaction_time
     if interaction_timeout and not self._prev_timed_out:
-      for callback in self._interactive_timeout_callbacks:
-        callback()
+      self._interactive_timeout_callbacks()
     self._prev_timed_out = interaction_timeout
 
     self._set_awake(ui_state.ignition or not interaction_timeout or PC)

system/ui/lib/callback_manager.py

@@ -0,0 +1,68 @@
+import weakref
+from collections.abc import Callable
+from typing import Any
+
+
+class CallbackManager:
+  def __init__(self):
+    self._callbacks: weakref.WeakSet[weakref.WeakMethod] = weakref.WeakSet()
+
+  def add(self, callback_method: Callable[..., Any]) -> None:
+    weak_cb = weakref.WeakMethod(callback_method)
+    self._callbacks.add(weak_cb)
+
+  def remove(self, callback_method: Callable[..., Any]) -> None:
+    weak_cb = weakref.WeakMethod(callback_method)
+    self._callbacks.discard(weak_cb)
+
+  def clear(self) -> None:
+    self._callbacks.clear()
+
+  def __call__(self, *args, **kwargs) -> None:
+    for weak_cb in list(self._callbacks):
+      callback = weak_cb()
+      if callback is not None:
+        # The listener instance is still alive, call the method
+        callback(*args, **kwargs)
+      else:
+        # The listener instance is gone. WeakSet handles automatic removal,
+        pass
+
+
+class SingleCallback:
+  def __init__(self, callback: Callable[..., Any] | None = None):
+    self._callback: weakref.WeakMethod | Callable | None = None
+    if callback is not None:
+      self.set(callback)
+
+  def set(self, callback: Callable[..., Any] | None) -> None:
+    """Set the callback. Supports bound methods, functions, and lambdas."""
+    if callback is None:
+      self._callback = None
+    elif hasattr(callback, '__self__'):
+      # Bound method - use WeakMethod to avoid keeping instance alive
+      self._callback = weakref.WeakMethod(callback)
+    elif callable(callback):
+      # Lambdas and regular functions - store directly (strong reference)
+      self._callback = callback
+    else:
+      raise TypeError(f"Expected callable, got {type(callback)}")
+
+  def clear(self) -> None:
+    self._callback = None
+
+  def __call__(self, *args, **kwargs) -> Any:
+    if self._callback is None:
+      return None
+
+    if isinstance(self._callback, weakref.WeakMethod):
+      callback = self._callback()
+      if callback is not None:
+        return callback(*args, **kwargs)
+      else:
+        # Instance was garbage collected
+        self._callback = None
+        return None
+    else:
+      # Direct callable reference (lambda or function)
+      return self._callback(*args, **kwargs)

system/ui/widgets/__init__.py

@@ -4,6 +4,7 @@
 from collections.abc import Callable
 from openpilot.common.filter_simple import BounceFilter, FirstOrderFilter
 from openpilot.system.ui.lib.application import gui_app, MousePos, MAX_TOUCH_SLOTS, MouseEvent
+from openpilot.system.ui.lib.callback_manager import SingleCallback
 
 try:
   from openpilot.selfdrive.ui.ui_state import device
@@ -230,7 +231,7 @@ class NavWidget(Widget, abc.ABC):
 
   def __init__(self):
     super().__init__()
-    self._back_callback: Callable[[], None] | None = None
+    self._back_callback: SingleCallback | None = None
     self._back_button_start_pos: MousePos | None = None
     self._swiping_away = False  # currently swiping away
     self._can_swipe_away = True  # swipe away is blocked after certain horizontal movement
@@ -253,7 +254,7 @@ def set_back_enabled(self, enabled: bool | Callable[[], bool]) -> None:
     self._back_enabled = enabled
 
   def set_back_callback(self, callback: Callable[[], None]) -> None:
-    self._back_callback = callback
+    self._back_callback = SingleCallback(callback)
 
   def _handle_mouse_event(self, mouse_event: MouseEvent) -> None:
     super()._handle_mouse_event(mouse_event)