From 6bc930a07b13f968c0384b0ec162debea077b00c Mon Sep 17 00:00:00 2001 From: firestarsdog <229254897+firestarsdog@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:14:25 -0400 Subject: [PATCH] Color border cleanup --- openpilot/selfdrive/ui/ui_state.py | 10 +++- openpilot/starpilot/ui/runtime_snapshot.py | 8 ++- .../starpilot/ui/tests/test_onroad_axes.py | 52 +++++++++++++++++++ .../ui/tests/test_ui_started_authority.py | 26 ++++++++++ 4 files changed, 92 insertions(+), 4 deletions(-) diff --git a/openpilot/selfdrive/ui/ui_state.py b/openpilot/selfdrive/ui/ui_state.py index d360c496ec..573ad5c7cf 100644 --- a/openpilot/selfdrive/ui/ui_state.py +++ b/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) diff --git a/openpilot/starpilot/ui/runtime_snapshot.py b/openpilot/starpilot/ui/runtime_snapshot.py index f40c497105..44c9d40024 100644 --- a/openpilot/starpilot/ui/runtime_snapshot.py +++ b/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( diff --git a/openpilot/starpilot/ui/tests/test_onroad_axes.py b/openpilot/starpilot/ui/tests/test_onroad_axes.py index 01673d005b..a84017cfe8 100644 --- a/openpilot/starpilot/ui/tests/test_onroad_axes.py +++ b/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 diff --git a/openpilot/starpilot/ui/tests/test_ui_started_authority.py b/openpilot/starpilot/ui/tests/test_ui_started_authority.py index 6d6cb007ec..8ba217ecf4 100644 --- a/openpilot/starpilot/ui/tests/test_ui_started_authority.py +++ b/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():