diff --git a/openpilot/starpilot/ui/runtime_snapshot.py b/openpilot/starpilot/ui/runtime_snapshot.py index f9d42a2e3d..15bbde1262 100644 --- a/openpilot/starpilot/ui/runtime_snapshot.py +++ b/openpilot/starpilot/ui/runtime_snapshot.py @@ -399,8 +399,11 @@ class RuntimeSnapshotAdapter: mono_clock=self._mono_clock, boot_clock=self._boot_clock) self._replay_epoch = replay.epoch now_ns = replay.now_ns if replay is not None and replay.valid else 0 + traffic_clock = (replay.now_ns, replay.boot_ns) if replay is not None and replay.valid and replay.boot_ns is not None else None else: now_ns = time.monotonic_ns() if now_ns is None else now_ns + # Pair the Traffic clocks before settings reads and snapshot work can delay them. + traffic_clock = paired_clocks_ns() if mode != ShellMode.ONROAD or self._last_mode != ShellMode.ONROAD: self._onroad_ancillary = None self._onroad_view_ancillary = None @@ -546,14 +549,10 @@ class RuntimeSnapshotAdapter: except (AttributeError, OSError, TypeError, ValueError): self._traffic_profile_valid = False self._traffic_context_ns = now_ns - if getattr(ui, "replay_clock", None) is not None: - pair = (replay.now_ns, replay.boot_ns) if replay is not None and replay.valid and replay.boot_ns is not None else None - else: - pair = paired_clocks_ns() - if pair is not None and abs(pair[0] - now_ns) <= 10_000_000: + if traffic_clock is not None and abs(traffic_clock[0] - now_ns) <= 10_000_000: fingerprint = settings_fingerprint(self._traffic_settings.refresh(now_ns)) traffic = self._traffic.project( - traffic_envelope, now_mono_ns=now_ns, now_boot_ns=pair[1], + traffic_envelope, now_mono_ns=now_ns, now_boot_ns=traffic_clock[1], drive_id=drive_id, settings_fingerprint=fingerprint, map_fingerprint=self._traffic_map_fingerprint, map_assigned=self._traffic_map_assigned, diff --git a/openpilot/starpilot/ui/tests/test_onroad_axes.py b/openpilot/starpilot/ui/tests/test_onroad_axes.py index 1a4d5a8f92..2b0da06ec3 100644 --- a/openpilot/starpilot/ui/tests/test_onroad_axes.py +++ b/openpilot/starpilot/ui/tests/test_onroad_axes.py @@ -281,8 +281,17 @@ class TestOnroadAxes(unittest.TestCase): (OnroadAlert(size=AlertSize.SMALL, text1='Steering Paused'), (10, 186, 181, 255)), (OnroadAlert(size=AlertSize.MID), (10, 186, 181, 255)), (OnroadAlert(size=AlertSize.FULL), blue), (OnroadAlert(size=AlertSize.SMALL, critical=True), blue))] - for lateral, longitudinal, override, alert, compact_color, latched in cases: - state = replace(self.state(lateral, longitudinal), driver_override_active=override, alert=alert, aol_latched=latched) + cases = [(*case, False) for case in cases] + cases += [(True, True, override, alert, color, None, True) for override, alert, color in ( + (False, OnroadAlert(), (201, 34, 49, 255)), + (False, OnroadAlert(size=AlertSize.SMALL, critical=True), blue), + (False, OnroadAlert(), (201, 34, 49, 255)), + (True, OnroadAlert(), gray), + (False, OnroadAlert(size=AlertSize.FULL), blue), + (False, OnroadAlert(), (201, 34, 49, 255)))] + for lateral, longitudinal, override, alert, compact_color, latched, traffic in cases: + state = replace(self.state(lateral, longitudinal), driver_override_active=override, alert=alert, + aol_latched=latched, traffic_mode=traffic) for layout in state.customization['layouts'].values(): layout[MODE_WIDGET]['enabled'] = True for profile in (Profile.LARGE, Profile.COMPACT): diff --git a/openpilot/starpilot/ui/tests/test_traffic_status.py b/openpilot/starpilot/ui/tests/test_traffic_status.py index 3f707f155a..22387165c1 100644 --- a/openpilot/starpilot/ui/tests/test_traffic_status.py +++ b/openpilot/starpilot/ui/tests/test_traffic_status.py @@ -16,7 +16,7 @@ from openpilot.starpilot.conditional_mode.manual import read_button_map from openpilot.starpilot.conditional_mode.runtime_settings import ConditionalSettingsOwner from openpilot.starpilot.conditional_mode.status import settings_fingerprint from openpilot.starpilot.ui.onroad_compact_widgets import MiciSidebarWidgets -from openpilot.starpilot.ui.onroad import OnroadView +from openpilot.starpilot.ui.onroad import OnroadView, axis_status_color from openpilot.starpilot.ui.presentation import FontRole, Profile from openpilot.starpilot.ui.runtime_snapshot import RuntimeSnapshotAdapter from openpilot.starpilot.ui.shell import ShellMode @@ -110,6 +110,55 @@ class TestTrafficStatus(unittest.TestCase): self.assertIsNone(self.project(first)) self.assertIsNone(self.project(next_frame, drive_id=DRIVE + 1)) + def test_snapshot_work_does_not_blink_fresh_traffic_border(self): + ui = ui_fake() + ui.params = self.params + ui.CP = interfaces[CAR.HYUNDAI_IONIQ_6].get_non_essential_params(CAR.HYUNDAI_IONIQ_6) + ui.CP.flags = int(HyundaiFlags.CANFD_LKA_STEER_MSG) + ui.CP.openpilotLongitudinalControl = True + ui.sm['deviceState'].startedMonoTime = DRIVE + ui.sm['carState'].canValid = True + ui.sm['carState'].canTimeout = False + ui.sm['carControl'].longActive = True + ui.sm['selfdriveState'].enabled = True + ui.sm.put('controlsState', NS(longControlState='pid')) + adapter = RuntimeSnapshotAdapter(ui, mono_clock=lambda: NOW, boot_clock=lambda: NOW + BOOT_OFFSET_NS) + clock = [NOW] + # A delayed settings read must not move clock acquisition away from the snapshot. + from openpilot.starpilot.ui import runtime_snapshot + read_settings = runtime_snapshot.read_traffic_settings + + def delayed_settings(params): + clock[0] += 30_000_000 + return read_settings(params) + + with mock.patch.object(runtime_snapshot, 'paired_clocks_ns', side_effect=lambda: (clock[0], clock[0] + BOOT - NOW, 0)), \ + mock.patch.object(runtime_snapshot, 'read_traffic_settings', side_effect=delayed_settings): + for frame, delay_read in enumerate((True, False, True, False, True)): + now = NOW + frame * 10_000_000 + clock[0] = now + if delay_read: + adapter._traffic_context_ns = None + event = receipt(self.settings_hash, self.map_hash, observed=now, sequence=frame + 1) + ui.sm.put('slcState', event.slcState) + for _ in range(2): + clock[0] = now + shown = adapter.build(ShellMode.ONROAD, now_ns=now).onroad + self.assertTrue(shown.traffic_mode) + color = axis_status_color(shown) + self.assertEqual((color.r, color.g, color.b, color.a), (201, 34, 49, 255)) + # Actual expiry still clears status; no cached red border or grace period. + clock[0] = now + 100_000_001 + shown = adapter.build(ShellMode.ONROAD, now_ns=clock[0]).onroad + self.assertFalse(shown.traffic_mode) + self.assertIsNone(shown.traffic_display) + # A mismatched or unavailable clock pair must still fail closed. + clock[0] = now + self.assertTrue(adapter.build(ShellMode.ONROAD, now_ns=now).onroad.traffic_mode) + for pair in (None, (now + 10_000_001, BOOT, 0)): + with mock.patch.object(runtime_snapshot, 'paired_clocks_ns', return_value=pair): + self.assertFalse(adapter.build(ShellMode.ONROAD, now_ns=now).onroad.traffic_mode) + def test_snapshot_and_compact_rail_use_only_live_effective_status(self): ui = ui_fake() ui.params = self.params