Color border cleanup

openpilot/selfdrive/ui/ui_state.py

@@ -12,6 +12,7 @@ from opendbc.car.structs import car
 from openpilot.common.filter_simple import FirstOrderFilter
 from openpilot.common.params import Params
 from openpilot.starpilot.schema_cache import get_cache, prewarm_cache_contracts
+from openpilot.starpilot.feature_runtime import enabled as feature_enabled
 from openpilot.starpilot.ui.display_preferences import AUTO, DisplayPreferences, read_preferences
 from openpilot.common.realtime import drop_realtime
 from openpilot.common.swaglog import cloudlog
@@ -128,6 +129,7 @@ class UIState:
         "navRoute",
         "carOutput",
         "carControl",
+        "aolAxisState",
         "slcState",
         "vehicleParameters",
         "testJoystick",
@@ -135,8 +137,9 @@ class UIState:
         *(["customReserved6", "cabinCameraState", "narrowRoadEncodeIdx", "wideRoadEncodeIdx", "cabinEncodeIdx"]
           if self.replay_clock is not None else []),
       ],
-      ignore_alive=["starpilotSelfdriveState", "starpilotLateralState"], ignore_valid=["starpilotSelfdriveState", "starpilotLateralState"],
-      ignore_avg_freq=["starpilotSelfdriveState", "starpilotLateralState"],
+      ignore_alive=["starpilotSelfdriveState", "starpilotLateralState", "aolAxisState"],
+      ignore_valid=["starpilotSelfdriveState", "starpilotLateralState", "aolAxisState"],
+      ignore_avg_freq=["starpilotSelfdriveState", "starpilotLateralState", "aolAxisState"],
     )
 
     self.prime_state = ProjectionPrimeState(self.params) if self.replay_clock is not None or isinstance(self.params, ProjectionParams) else PrimeState()
@@ -176,6 +179,7 @@ class UIState:
     self.panda_type: log.PandaState.PandaType = log.PandaState.PandaType.unknown
     self.personality: log.LongitudinalPersonality = log.LongitudinalPersonality.standard
     self.has_longitudinal_control: bool = False
+    self.aol_runtime: bool = False
     self.is_body: bool | None = False
     self.CP: car.CarParams | None = None
     self.light_sensor: float = -1.0
@@ -302,6 +306,7 @@ class UIState:
     self.started_frame = max(0, self.sm.frame - 1)
     self.CP = None
     self.has_longitudinal_control = False
+    self.aol_runtime = False
     self.ignition = False
     self.light_sensor = -1.0
     self._projection_params_at = None
@@ -434,6 +439,7 @@ class UIState:
       self.CP = None
       self.has_longitudinal_control = False
 
+    self.aol_runtime = bool(self.CP is not None and feature_enabled(self.params, self.CP, "aol", os.environ))
     self.recording_audio = self.params.get_bool("RecordAudio") and self.started
     self.is_metric = self.params.get_bool("IsMetric")
     self.initial_cruise_speed = self.params.get("InitialCruiseSpeedKph", return_default=True)

openpilot/starpilot/ui/runtime_snapshot.py

@@ -21,6 +21,7 @@ from openpilot.starpilot.curve_speed.status import observation as curve_observat
 from openpilot.starpilot.conditional_mode.policy import ModeChoice
 from openpilot.starpilot.conditional_mode.manual import TRAFFIC_MODE_ACTION, read_button_map
 from openpilot.starpilot.feature_capabilities import lateral_supported, longitudinal_supported
+from openpilot.starpilot.aol.runtime import current_axis
 from openpilot.starpilot.conditional_mode.projection import paired_clocks_ns
 from openpilot.starpilot.conditional_mode.runtime_settings import ConditionalSettingsOwner
 from openpilot.starpilot.conditional_mode.status import settings_fingerprint
@@ -631,9 +632,12 @@ class RuntimeSnapshotAdapter:
     )
     display_lat_active = bool(display_control.latActive) if display_control is not None else False
     display_long_active = bool(display_control.longActive) if display_control is not None else False
+    long_engaged = bool(display_selfdrive is not None and getattr(display_selfdrive, 'enabled', False))
+    if getattr(ui, 'aol_runtime', False):
+      axis = current_axis(sm, now_ns=now_ns) if started and sm.recv_frame.get('aolAxisState', 0) > after else None
+      long_engaged = long_engaged and bool(axis is not None and axis.nativeAcknowledged and axis.desiredLongitudinal)
     longitudinal_overridden = bool(
-      display_car_valid and getattr(display_car, 'gasPressed', False) and
-      display_selfdrive is not None and getattr(display_selfdrive, 'enabled', False) and
+      display_car_valid and getattr(display_car, 'gasPressed', False) and long_engaged and
       str(getattr(display_selfdrive, 'state', '')) == 'overriding' and
       cp is not None and cp.openpilotLongitudinalControl)
     developer_metrics = self._developer_metrics.observe(

openpilot/starpilot/ui/tests/test_onroad_axes.py

@@ -126,6 +126,58 @@ class TestOnroadAxes(unittest.TestCase):
         car.canValid, car.canTimeout = True, False
         setattr(car, field, False)
 
+  def test_independent_long_request_distinguishes_pause_from_pedal_override(self):
+    self.ui.aol_runtime = True
+    self.ui.sm['selfdriveState'].state = 'overriding'
+    axis = NS(qualified=True, sessionId='ui-axis', observedMonoTime=NOW, validUntilMonoTime=NOW + 30_000_000,
+              nativeAcknowledged=True, desiredLongitudinal=False)
+    self.ui.sm.put('aolAxisState', axis)
+    car = self.ui.sm['carState']
+    for requested, gas, steering, gray, pedal in (
+        (False, False, False, False, False), (False, True, False, False, False),
+        (False, False, True, True, False), (False, True, True, True, False),
+        (True, False, False, False, False), (True, True, False, True, True),
+        (True, False, True, True, False), (True, True, True, True, True)):
+      with self.subTest(requested=requested, gas=gas, steering=steering):
+        axis.desiredLongitudinal = requested
+        car.gasPressed, car.steeringPressed = gas, steering
+        state = self.state(True, False)
+        self.assertEqual(state.longitudinal_overridden, pedal)
+        self.assertEqual(state.driver_override_active, gray)
+        self.assertEqual(state.cruise_active, pedal)
+        self.assertEqual(driving_mode_description(state), 'ACC - Override' if pedal else 'Always On Lateral')
+        self.assertEqual(rgba(axis_status_color(state)), (145, 155, 149, 255) if gray else (10, 186, 181, 255))
+
+  def test_independent_pedal_override_requires_current_axis_request(self):
+    self.ui.aol_runtime = True
+    self.ui.sm['selfdriveState'].state = 'overriding'
+    self.ui.sm['carState'].gasPressed = True
+    self.assertFalse(self.state(True, False).longitudinal_overridden)
+    axis = NS(qualified=True, sessionId='ui-axis', observedMonoTime=NOW, validUntilMonoTime=NOW + 30_000_000,
+              nativeAcknowledged=True, desiredLongitudinal=True)
+    self.ui.sm.put('aolAxisState', axis)
+    self.assertTrue(self.state(True, False).longitudinal_overridden)
+    for field, value in (('qualified', False), ('sessionId', ''), ('observedMonoTime', NOW + 1),
+                         ('validUntilMonoTime', NOW - 1), ('nativeAcknowledged', False)):
+      with self.subTest(field=field):
+        original = getattr(axis, field)
+        setattr(axis, field, value)
+        state = self.state(True, False)
+        self.assertFalse(state.longitudinal_overridden or state.driver_override_active or state.cruise_active)
+        setattr(axis, field, original)
+    for stamp in (NOW + 1, NOW - 31_000_000):
+      self.ui.sm.logMonoTime['aolAxisState'] = stamp
+      self.assertFalse(self.state(True, False).longitudinal_overridden)
+    self.ui.sm.logMonoTime['aolAxisState'] = NOW
+    self.ui.sm.recv_frame['aolAxisState'] = self.ui.started_frame
+    self.assertFalse(self.state(True, False).longitudinal_overridden)
+    self.ui.sm.recv_frame['aolAxisState'] = self.ui.started_frame + 1
+    for flags in (self.ui.sm.valid, self.ui.sm.alive, self.ui.sm.seen):
+      flags['aolAxisState'] = False
+      self.assertFalse(self.state(True, False).longitudinal_overridden)
+      flags['aolAxisState'] = True
+    self.assertTrue(self.state(True, False).longitudinal_overridden)
+
   def test_traffic_returns_after_conditional_stop_without_changing_intent(self):
     from openpilot.starpilot.conditional_mode.policy import ModeChoice
     from openpilot.starpilot.ui.conditional_status import ConditionalDisplay

openpilot/starpilot/ui/tests/test_ui_started_authority.py

@@ -38,6 +38,30 @@ def test_replay_messages_use_clock_sample_after_receipt():
   ui._reset_replay_state.assert_not_called()
 
 
+@pytest.mark.parametrize('enabled', [False, True])
+def test_ui_reads_same_aol_runtime_admission_as_control_owner(enabled):
+  from openpilot.starpilot.ui.tests.test_runtime_snapshot import SubMasterFake
+
+  ui: Any = object.__new__(UIState)
+  ui.params = MagicMock()
+  ui.params.get.return_value = None
+  ui.params.get_bool.return_value = False
+  ui.replay_clock = object()
+  ui.sm = SubMasterFake()
+  cp = NS(openpilotLongitudinalControl=True)
+  ui.sm.put('carParams', cp)
+  ui.started = ui.chestnut_output_seen = False
+  with patch('openpilot.selfdrive.ui.ui_state.get_cache', return_value=None), \
+       patch('openpilot.selfdrive.ui.ui_state.feature_enabled', return_value=enabled) as admission:
+    ui.update_params()
+    assert ui.aol_runtime is enabled
+    assert admission.call_args.args[:3] == (ui.params, cp, 'aol')
+    ui.sm.seen['carParams'] = False
+    ui.update_params()
+    assert not ui.aol_runtime
+    admission.assert_called_once()
+
+
 def owner():
   ui = object.__new__(UIState)
   messages = {'deviceState': NS(started=False, chestnutPresent=False),
@@ -121,6 +145,7 @@ def test_started_presentation_does_not_replace_existing_message_freshness():
 def replay_owner(sample):
   from openpilot.starpilot.ui.tests.test_runtime_snapshot import SubMasterFake
   ui: Any = object.__new__(UIState)
+  ui.aol_runtime = True
   ui.sm = SubMasterFake()
   ui.sm.sock = {name: object() for name in ('carControl', 'carState', 'selfdriveState')}
   ui.sm.static_freq_services = set(ui.sm.sock)
@@ -154,6 +179,7 @@ def test_epoch_reset_precedes_paused_primer_and_never_drains_socket():
   drain.assert_not_called()
   assert display_message(ui.sm, 'carControl', NOW, after_frame=ui.started_frame).latActive
   assert ui.sm.alive['carControl']
+  assert not ui.aol_runtime
 
 
 def test_straddled_epoch_withholds_frame_but_retains_new_paused_primer():