diff --git a/openpilot/common/hardware/pc/hardware.py b/openpilot/common/hardware/pc/hardware.py index 417d2db37d..a2c18240f2 100644 --- a/openpilot/common/hardware/pc/hardware.py +++ b/openpilot/common/hardware/pc/hardware.py @@ -1,8 +1,15 @@ +import os +import re + from openpilot.cereal import log from openpilot.common.hardware.base import HardwareBase class HardwarePc(HardwareBase): def get_device_type(self): + recorded = os.getenv("SP_REPLAY_DEVICE_TYPE", "") + if (os.getenv("SP_HOST_RUNTIME") == "1" and recorded in ("tici", "tizi", "mici") and + re.fullmatch(r"replay-[A-Za-z0-9_-]{1,48}", os.getenv("OPENPILOT_PREFIX", ""))): + return recorded return "pc" def get_network_type(self): diff --git a/openpilot/selfdrive/ui/ui.py b/openpilot/selfdrive/ui/ui.py index 1ddb883147..8f2069b408 100755 --- a/openpilot/selfdrive/ui/ui.py +++ b/openpilot/selfdrive/ui/ui.py @@ -44,10 +44,10 @@ def main(): borrowed_messages=True, evidence_wait_ms=0) preview_authority = LiveContextSource(ui_state.params, messages=ui_state.sm, borrowed_messages=True, evidence_wait_ms=0) - controllers = ControllerRuntime(ui_state.params, actions=layout.star._favorite_actions, + controllers = None if ui_state.replay_clock is not None else ControllerRuntime(ui_state.params, actions=layout.star._favorite_actions, favorites=layout.star.favorites_owner.snapshot, invoke_favorite=layout.star.favorites_owner.invoke, authority=controller_authority) - preview = LayoutPreviewRuntime(ui_state.params, (Profile.LARGE if BIG_UI else Profile.COMPACT).value, + preview = None if ui_state.replay_clock is not None else LayoutPreviewRuntime(ui_state.params, (Profile.LARGE if BIG_UI else Profile.COMPACT).value, offroad_hint=ui_state.is_offroad, authority=preview_authority) else: layout = MainLayout() if BIG_UI else MiciMainLayout() @@ -84,7 +84,8 @@ def main(): preview.close() finally: try: - controllers.close() + if controllers is not None: + controllers.close() finally: layout.close() diff --git a/openpilot/selfdrive/ui/ui_state.py b/openpilot/selfdrive/ui/ui_state.py index 5234197034..c59f4f151b 100644 --- a/openpilot/selfdrive/ui/ui_state.py +++ b/openpilot/selfdrive/ui/ui_state.py @@ -2,6 +2,7 @@ import numpy as np import os import time import threading +import capnp from collections.abc import Callable from enum import Enum from openpilot.cereal import messaging, log @@ -17,6 +18,7 @@ from openpilot.system.ui.lib.application import gui_app from openpilot.common.hardware import HARDWARE, PC from openpilot.common.hardware.usb import cable_connected, get_usb_state, is_chestnut_usb_id from openpilot.starpilot.models.manager import ModelManager +from openpilot.tools.replay.display_clock import DisplayClockReader BACKLIGHT_OFFROAD = 65 if HARDWARE.get_device_type() == "mici" else 50 PARAM_UPDATE_TIME = 1 / 5.0 @@ -86,6 +88,8 @@ class UIState: def _initialize(self): prewarm_cache_contracts() + self.replay_clock = DisplayClockReader() if DisplayClockReader.enabled() else None + self.replay_sample = None self.projection_read_only = os.environ.get("STARPILOT_PROJECTION_READ_ONLY") == "1" params = Params() self.params = ProjectionParams(params) if self.projection_read_only else params @@ -124,7 +128,7 @@ class UIState: ignore_avg_freq=["starpilotSelfdriveState", "starpilotLateralState"], ) - self.prime_state = ProjectionPrimeState(self.params) if isinstance(self.params, ProjectionParams) else PrimeState() + self.prime_state = ProjectionPrimeState(self.params) if self.replay_clock is not None or isinstance(self.params, ProjectionParams) else PrimeState() # UI Status tracking self.status: UIStatus = UIStatus.DISENGAGED @@ -139,7 +143,7 @@ class UIState: self.always_on_dm: bool = self.params.get_bool("AlwaysOnDM") self.experimental_mode: bool = self.params.get_bool("ExperimentalMode") self.chestnut_present: bool = False - self._gpu_artifacts = ModelManager() + self._gpu_artifacts = None if self.replay_clock is not None else ModelManager() self._gpu_artifacts_at: float | None = None self.chestnut_compiled: bool = False self.chestnut_checking: bool = False @@ -189,6 +193,21 @@ class UIState: return not self.started def update(self) -> None: + if getattr(self, "replay_clock", None) is not None: + sample = self.replay_clock.sample() + if sample.epoch is not None and (self.replay_sample is None or sample.epoch != self.replay_sample.epoch): + self._reset_replay_state() + self.replay_sample = sample + self.sm.replay_sample = sample + self.sm.update(0) + self._update_state() + self._update_status() + now = time.monotonic() + if self._projection_params_at is None or now - self._projection_params_at >= 1.0: + self.update_params() + self._projection_params_at = now + self._update_chestnut_state() + return if self.projection_read_only: self.sm.update(0) self._update_state() @@ -209,6 +228,29 @@ class UIState: self._update_chestnut_state() device.update() + def display_time_ns(self) -> int: + if getattr(self, "replay_clock", None) is not None: + return self.replay_sample.now_ns if self.replay_sample is not None and self.replay_sample.valid else 0 + return time.monotonic_ns() + + def _reset_replay_state(self) -> None: + for service, socket in self.sm.sock.items(): + messaging.drain_sock_raw(socket) + for field in ("seen", "updated", "alive", "valid"): + getattr(self.sm, field)[service] = False + for field in ("recv_frame", "recv_time", "logMonoTime"): + getattr(self.sm, field)[service] = 0 + try: + empty = messaging.new_message(service) + except capnp.lib.capnp.KjException: + empty = messaging.new_message(service, 0) + self.sm.data[service] = getattr(empty.as_reader(), service) + self.started = self._started_prev = self._engaged_prev = False + self.started_frame = self.sm.frame + self.chestnut_output_seen = False + self.chestnut_present = False + self.status = UIStatus.DISENGAGED + def _params_refresh_worker(self): drop_realtime() while True: @@ -278,7 +320,7 @@ class UIState: if (self.started and self.sm.updated["selfdriveState"] and self.sm.valid["selfdriveState"] and self.sm.alive["selfdriveState"] and self.sm.recv_frame["selfdriveState"] > self.started_frame): stamp = self.sm.logMonoTime["selfdriveState"] - if 0 < stamp <= time.monotonic_ns() <= stamp + 250_000_000: + if 0 < stamp <= UIState.display_time_ns(self) <= stamp + 250_000_000: self.personality = log.LongitudinalPersonality.schema.enumerants[self.sm["selfdriveState"].personality] def _update_chestnut_state(self) -> None: @@ -293,7 +335,9 @@ class UIState: model_seen = self.sm.recv_frame["modelV2"] > self.started_frame model_stamp = self.sm.logMonoTime["modelV2"] model_fresh = (model_seen and self.sm.valid["modelV2"] and self.sm.alive["modelV2"] and - 0 < model_stamp <= time.monotonic_ns() <= model_stamp + 250_000_000) + 0 < model_stamp <= UIState.display_time_ns(self) <= model_stamp + 250_000_000) + if getattr(self, "replay_clock", None) is not None: + self.chestnut_present = detected if model_fresh and self.sm["modelV2"].big: self.chestnut_output_seen = True if not self.chestnut_present: @@ -315,7 +359,11 @@ class UIState: # For slower operations # Update longitudinal control state CP_bytes = get_cache(self.params, "CarParamsPersistent") - if CP_bytes is not None: + if (getattr(self, "replay_clock", None) is not None and + self.sm.seen["carParams"] and self.sm.valid["carParams"]): + self.CP = self.sm["carParams"] + self.has_longitudinal_control = self.CP.openpilotLongitudinalControl + elif CP_bytes is not None: self.CP = messaging.log_from_bytes(CP_bytes, car.CarParams) # The saved Alpha Long choice can survive an unavailable configuration. self.has_longitudinal_control = self.CP.openpilotLongitudinalControl @@ -331,6 +379,9 @@ class UIState: self.chestnut_active = raw_chestnut if type(raw_chestnut) is bool else ( True if raw_chestnut in (b"1", "1") else False if raw_chestnut in (b"0", "0") else None) self.chestnut_loading = self.params.get_bool("ChestnutLoading") + if getattr(self, "replay_clock", None) is not None: + self.chestnut_compiled = self.chestnut_active is True or self.chestnut_output_seen + return now = time.monotonic() if not self.started and (self._gpu_artifacts_at is None or now < self._gpu_artifacts_at or now - self._gpu_artifacts_at >= 1.): self.chestnut_compiled, self.chestnut_checking = self._gpu_artifacts.selected_gpu_status() diff --git a/openpilot/starpilot/ui/runtime_app.py b/openpilot/starpilot/ui/runtime_app.py index d5f1d4c81b..32fdd28da6 100644 --- a/openpilot/starpilot/ui/runtime_app.py +++ b/openpilot/starpilot/ui/runtime_app.py @@ -268,6 +268,8 @@ class StarShellSession: self._favorite_claimed = False def _favorite_authority(self) -> bool: + if getattr(ui_state, "replay_clock", None) is not None: + return False if self._mode != ShellMode.ONROAD: return False state = self.snapshot(ShellMode.ONROAD).onroad @@ -425,6 +427,8 @@ class StarShellSession: return actions def _poll_wheel(self, now_ns): + if getattr(ui_state, "replay_clock", None) is not None: + return for _ in range(32): event = messaging.recv_one_or_none(self._wheel_sock) if event is None: diff --git a/openpilot/starpilot/ui/runtime_snapshot.py b/openpilot/starpilot/ui/runtime_snapshot.py index a0e39d6182..d6ec8c4ed0 100644 --- a/openpilot/starpilot/ui/runtime_snapshot.py +++ b/openpilot/starpilot/ui/runtime_snapshot.py @@ -67,12 +67,15 @@ def _clock_pair(mono_clock: Callable[[], int], boot_clock: Callable[[], int]) -> def _message_at_age(sm: Any, service: str, now_ns: int, after_frame: int, max_age_ns: int, *, recv_now_ns: int | None = None) -> Any | None: try: + replay = getattr(sm, "replay_sample", None) + if replay is not None and not replay.valid: + return None stamp = int(sm.logMonoTime[service]) if (not sm.valid[service] or not sm.alive[service] or stamp <= 0 or stamp > now_ns or now_ns - stamp > max_age_ns or int(sm.recv_frame[service]) <= after_frame): return None - if recv_now_ns is not None: - receipt_age = recv_now_ns - int(sm.recv_time[service] * 1e9) + if recv_now_ns is not None and not (replay is not None and replay.paused): + receipt_age = (replay.host_ns if replay is not None else recv_now_ns) - int(sm.recv_time[service] * 1e9) if not 0 <= receipt_age <= max_age_ns: return None return sm[service] @@ -87,6 +90,11 @@ def current_message(sm: Any, service: str, now_ns: int, *, after_frame: int = 0, max_age_ns = int(2e9 / frequency) if frequency else 1_000_000_000 if service == "pandaStates": if boot_now_ns is None: + replay = getattr(sm, "replay_sample", None) + if replay is not None: + if not replay.valid or replay.boot_ns is None: + return None + return _message_at_age(sm, service, replay.boot_ns, after_frame, max_age_ns, recv_now_ns=replay.host_ns) pair = _clock_pair(time.monotonic_ns, _boot_time_ns) if pair is None: return None @@ -103,6 +111,9 @@ def display_message(sm: Any, service: str, now_ns: int, *, after_frame: int = 0) def current_curve_message(sm: Any, now_ns: int, *, after_frame: int = 0) -> Any | None: """Read the nested Curve payload without treating SLC's Event.valid as its validity.""" try: + replay = getattr(sm, "replay_sample", None) + if replay is not None and not replay.valid: + return None stamp = int(sm.logMonoTime["slcState"]) frequency = SERVICE_LIST["slcState"].frequency max_age_ns = int(2e9 / frequency) if frequency else 1_000_000_000 @@ -190,10 +201,14 @@ def current_alert(sm: Any, now_ns: int, *, after_frame: int) -> OnroadAlert: service = 'selfdriveState' stamp = int(sm.logMonoTime[service]) receipt = int(sm.recv_time[service] * 1e9) - if (not sm.valid[service] or int(sm.recv_frame[service]) <= after_frame or - stamp <= 0 or stamp > now_ns or receipt <= 0 or receipt > now_ns): + replay = getattr(sm, "replay_sample", None) + if replay is not None and not replay.valid: return OnroadAlert() - missing_ns = now_ns - receipt + receipt_now = replay.host_ns if replay is not None else now_ns + if (not sm.valid[service] or int(sm.recv_frame[service]) <= after_frame or + stamp <= 0 or stamp > now_ns or receipt <= 0 or receipt > receipt_now): + return OnroadAlert() + missing_ns = 0 if replay is not None and replay.paused else receipt_now - receipt if missing_ns > _ALERT_TRANSPORT_TIMEOUT_NS: previous = sm[service] if bool(previous.enabled) and missing_ns < _ALERT_TRANSPORT_TIMEOUT_NS + _ALERT_CRITICAL_TIMEOUT_NS: @@ -282,6 +297,7 @@ class RuntimeSnapshotAdapter: self._device_offset_ns: int | None = None self._last_pair_offset_ns: int | None = None self._pair_failed = False + self._replay_epoch: int | None = None def _parked(self, ui: Any, pair: tuple[int, int, int] | None) -> bool: if pair is None: @@ -319,6 +335,8 @@ class RuntimeSnapshotAdapter: def confirmed_offroad(self) -> bool: """Recheck transport evidence without changing presentation state.""" + if getattr(self.ui_state, "replay_clock", None) is not None: + return False return self._parked(self.ui_state, _clock_pair(self._mono_clock, self._boot_clock)) def connectivity_allowed(self) -> bool: @@ -328,7 +346,15 @@ class RuntimeSnapshotAdapter: def build(self, mode: ShellMode, selected: Destination = Destination.STAR, *, compact_y: float = 0, compact_scroll_x: float = 0, sidebar_expanded: bool = True, now_ns: int | None = None, menu_only: bool = False) -> ShellSnapshot: - now_ns = time.monotonic_ns() if now_ns is None else now_ns + replay = getattr(self.ui_state, "replay_sample", None) + if getattr(self.ui_state, "replay_clock", None) is not None: + if replay is not None and replay.epoch != self._replay_epoch: + self.__init__(self.ui_state, self.galaxy_access, self.bluetooth_powered, + 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 + else: + now_ns = time.monotonic_ns() if now_ns is None else now_ns if mode != ShellMode.ONROAD or self._last_mode != ShellMode.ONROAD: self._onroad_ancillary = None if mode == ShellMode.ONROAD and self._last_mode == ShellMode.SETTINGS: @@ -347,13 +373,17 @@ class RuntimeSnapshotAdapter: self._conditional_read_ns = now_ns sm = ui.sm device = current_message(sm, "deviceState", now_ns) - pair = _clock_pair(self._mono_clock, self._boot_clock) + if getattr(ui, "replay_clock", None) is not None: + pair = ((replay.now_ns, replay.boot_ns, replay.boot_ns - replay.now_ns) + if replay is not None and replay.valid and replay.boot_ns is not None else None) + else: + pair = _clock_pair(self._mono_clock, self._boot_clock) pandas = current_message(sm, "pandaStates", now_ns, boot_now_ns=now_ns + pair[2]) if pair is not None else None panda_connected = bool(pandas and any(p.pandaType != log.PandaState.PandaType.unknown for p in pandas)) # UIState owns drive transitions. Health-message freshness governs actions, # not the lifetime of the camera, widget filters, and engagement animations. started = bool(ui.started) - confirmed_offroad = self._parked(ui, pair) + confirmed_offroad = self._parked(ui, pair) if getattr(ui, "replay_clock", None) is None else False after = ui.started_frame if ui.started else 0 drive_id = int(getattr(sm["deviceState"], "startedMonoTime", 0)) if started else 0 car = current_message(sm, "carState", now_ns, after_frame=after) if started else None @@ -445,7 +475,10 @@ class RuntimeSnapshotAdapter: except (AttributeError, OSError, TypeError, ValueError): self._traffic_profile_valid = False self._traffic_context_ns = now_ns - pair = paired_clocks_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: fingerprint = settings_fingerprint(self._traffic_settings.refresh(now_ns)) traffic = self._traffic.project( @@ -517,6 +550,11 @@ class RuntimeSnapshotAdapter: experimental = bool(display_selfdrive.experimentalMode) if display_selfdrive is not None else _flag(params, "ExperimentalMode") observed_experimental = bool(display_selfdrive.experimentalMode) if display_selfdrive is not None else False persona = _personality(_text(params, "LongitudinalPersonality", "1")) + if display_selfdrive is not None: + try: + persona = _personality(log.LongitudinalPersonality.schema.enumerants[str(display_selfdrive.personality)]) + except (AttributeError, KeyError, TypeError, ValueError): + pass torque = observe_torque_feedback( display_car, display_control, display_controls, display_message(sm, "carOutput", now_ns, after_frame=after) if started else None, diff --git a/openpilot/starpilot/ui/tests/test_host_launch.py b/openpilot/starpilot/ui/tests/test_host_launch.py index cbacd91303..ca7696fce6 100644 --- a/openpilot/starpilot/ui/tests/test_host_launch.py +++ b/openpilot/starpilot/ui/tests/test_host_launch.py @@ -80,3 +80,14 @@ class TestHostLaunch(unittest.TestCase): with patch.dict(os.environ, SP_C3_COMPILE_ONLY="1"), patch.object(host_launch, "seed_developer_defaults") as seed: self.assertEqual(host_launch.main(["large", "--any-ui-arg"]), 0) seed.assert_not_called() + + def test_recorded_hardware_identity_requires_private_host_replay(self): + from openpilot.common.hardware.pc.hardware import HardwarePc + hardware = HardwarePc() + with patch.dict(os.environ, SP_HOST_RUNTIME="1", SP_REPLAY_DEVICE_TYPE="tizi", OPENPILOT_PREFIX="replay-device"): + self.assertEqual(hardware.get_device_type(), "tizi") + os.environ["OPENPILOT_PREFIX"] = "d" + self.assertEqual(hardware.get_device_type(), "pc") + os.environ["OPENPILOT_PREFIX"] = "replay-device" + os.environ["SP_REPLAY_DEVICE_TYPE"] = "unknown" + self.assertEqual(hardware.get_device_type(), "pc") diff --git a/openpilot/starpilot/ui/tests/test_runtime_snapshot.py b/openpilot/starpilot/ui/tests/test_runtime_snapshot.py index 4641f461d7..b447cf747f 100644 --- a/openpilot/starpilot/ui/tests/test_runtime_snapshot.py +++ b/openpilot/starpilot/ui/tests/test_runtime_snapshot.py @@ -97,6 +97,63 @@ def ui_fake(): class TestRuntimeSnapshot(unittest.TestCase): + def test_replay_uses_recorded_time_without_rewriting_messages(self): + from openpilot.tools.replay.display_clock import ClockSample + ui = ui_fake() + ui.replay_clock = object() + ui.replay_sample = ClockSample(NOW, None, NOW * 1000, epoch=1, valid=True) + ui.sm.replay_sample = ui.replay_sample + ui.sm['selfdriveState'].personality = 'relaxed' + before = dict(ui.sm.logMonoTime) + state = NativeRuntimeSnapshotAdapter(ui).build(ShellMode.ONROAD).onroad + self.assertTrue(state.engaged) + self.assertEqual(state.personality, 2) + self.assertEqual(state.observed_ns, NOW) + self.assertEqual(ui.sm.logMonoTime, before) + ui.replay_sample = ClockSample(None, None, NOW * 1000, epoch=1) + ui.sm.replay_sample = ui.replay_sample + self.assertFalse(NativeRuntimeSnapshotAdapter(ui).build(ShellMode.ONROAD).onroad.engaged) + + def test_replay_pause_holds_visuals_but_publisher_death_does_not(self): + from openpilot.tools.replay.display_clock import ClockSample + ui = ui_fake() + ui.replay_clock = object() + ui.replay_sample = ClockSample(NOW, None, NOW * 1000, epoch=1, paused=True, valid=True) + ui.sm.replay_sample = ui.replay_sample + ui.sm['selfdriveState'].alertText1 = 'Recorded alert' + ui.sm['selfdriveState'].alertSize = NS(raw=1) + state = NativeRuntimeSnapshotAdapter(ui).build(ShellMode.ONROAD).onroad + self.assertTrue(state.engaged) + self.assertEqual(state.alert.text1, 'Recorded alert') + ui.sm.replay_sample = ClockSample(None, None, NOW * 1000, epoch=1, paused=True) + self.assertEqual(current_alert(ui.sm, NOW, after_frame=1).size, AlertSize.NONE) + + def test_replay_has_no_parked_control_authority(self): + from openpilot.tools.replay.display_clock import ClockSample + ui = ui_fake() + ui.replay_clock = object() + ui.started = False + ui.replay_sample = ClockSample(NOW, NOW, NOW * 1000, epoch=1, valid=True) + ui.sm.replay_sample = ui.replay_sample + adapter = NativeRuntimeSnapshotAdapter(ui) + self.assertFalse(adapter.confirmed_offroad()) + self.assertFalse(adapter.build(ShellMode.HOME).device.offroad) + + def test_replay_epoch_resets_projectors_and_saved_layout_cache(self): + from openpilot.tools.replay.display_clock import ClockSample + ui = ui_fake() + ui.replay_clock = object() + ui.replay_sample = ClockSample(NOW, None, NOW * 1000, epoch=1, valid=True) + ui.sm.replay_sample = ui.replay_sample + adapter = NativeRuntimeSnapshotAdapter(ui) + adapter.build(ShellMode.ONROAD) + previous = adapter._conditional + ui.replay_sample = ClockSample(NOW - 1_000_000_000, None, NOW * 1000, epoch=2, valid=True) + ui.sm.replay_sample = ui.replay_sample + state = adapter.build(ShellMode.ONROAD).onroad + self.assertIsNot(adapter._conditional, previous) + self.assertFalse(state.engaged) + def test_home_panda_presence_is_independent_of_ignition(self): for panda_type in (log.PandaState.PandaType.dos, log.PandaState.PandaType.tres, log.PandaState.PandaType.cuatro): for ignition in (False, True): diff --git a/openpilot/system/ui/lib/application.py b/openpilot/system/ui/lib/application.py index 2b89be995a..ac79ce8a8d 100644 --- a/openpilot/system/ui/lib/application.py +++ b/openpilot/system/ui/lib/application.py @@ -1039,6 +1039,9 @@ class GuiApplication: @staticmethod def big_ui() -> bool: + from openpilot.tools.replay.display_clock import DisplayClockReader + if DisplayClockReader.enabled(): + return BIG_UI return HARDWARE.get_device_type() in ('tici', 'tizi') or BIG_UI diff --git a/openpilot/tools/replay/display_clock.h b/openpilot/tools/replay/display_clock.h new file mode 100644 index 0000000000..59006a935d --- /dev/null +++ b/openpilot/tools/replay/display_clock.h @@ -0,0 +1,194 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "common/timing.h" + +class ReplayDisplayClock { +public: + static uint64_t host_now() { + timespec now; +#ifdef __APPLE__ + clock_gettime(CLOCK_UPTIME_RAW, &now); +#else + clock_gettime(CLOCK_MONOTONIC, &now); +#endif + return now.tv_sec * 1000000000ULL + now.tv_nsec; + } + static constexpr uint64_t MAGIC = 0x53505250434c4b31ULL; + static constexpr uint32_t VALID = 1, PAUSED = 2, SEEKING = 4, BOOT_KNOWN = 8; + struct Record { + uint64_t sequence, magic; + uint32_t version, flags; + uint64_t epoch, route_ns, host_ns; + double speed; + int64_t boot_offset_ns; + uint64_t heartbeat_ns, reserved; + }; + static_assert(sizeof(Record) == 80); + + ReplayDisplayClock() { + const char *runtime = std::getenv("SP_HOST_RUNTIME"); + const char *prefix_env = std::getenv("OPENPILOT_PREFIX"); + const char *path_env = std::getenv("SP_REPLAY_CLOCK_PATH"); + if (!runtime || std::string(runtime) != "1" || !prefix_env || !path_env) return; + std::string prefix(prefix_env); + if (prefix.rfind("replay-", 0) != 0 || prefix.size() <= 7 || prefix.size() > 55 || + prefix.find_first_not_of("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-") != std::string::npos) return; +#ifdef __APPLE__ + const std::string root = "/tmp"; +#else + const std::string root = "/dev/shm"; +#endif + const std::string parent = root + "/msgq_" + prefix; + if (std::string(path_env) != parent + "/display-clock") return; + int directory = open(parent.c_str(), O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + struct stat directory_stat = {}; + if (directory < 0) return; + if (fstat(directory, &directory_stat) || !S_ISDIR(directory_stat.st_mode) || + directory_stat.st_uid != getuid() || (directory_stat.st_mode & 07777) != 0700) { + close(directory); + return; + } + int file = openat(directory, "display-clock", O_RDWR | O_NONBLOCK | O_NOFOLLOW | O_CLOEXEC); + close(directory); + struct stat file_stat = {}; + if (file < 0) return; + if (fstat(file, &file_stat) || !S_ISREG(file_stat.st_mode) || file_stat.st_uid != getuid() || + (file_stat.st_mode & 07777) != 0600 || file_stat.st_nlink != 1 || file_stat.st_size != sizeof(Record) || flock(file, LOCK_EX | LOCK_NB)) { + close(file); + return; + } + void *mapping = mmap(nullptr, sizeof(Record), PROT_READ | PROT_WRITE, MAP_SHARED, file, 0); + if (mapping == MAP_FAILED) { + close(file); + return; + } + file_ = file; + shared_ = static_cast(mapping); + state_.magic = MAGIC; + state_.version = 1; + state_.epoch = 1; + state_.speed = 1.0; + state_.host_ns = host_now(); + write_locked(); + heartbeat_ = std::thread([this]() { + std::unique_lock lock(lock_); + while (!stopped_) { + if (wake_.wait_for(lock, std::chrono::milliseconds(50), [this]() { return stopped_; })) break; + write_locked(); + } + }); + } + + ~ReplayDisplayClock() { + stop(); + if (shared_) munmap(shared_, sizeof(Record)); + if (file_ >= 0) close(file_); + } + + void stop() { + { + std::lock_guard lock(lock_); + if (stopped_) return; + stopped_ = true; + state_.flags &= ~VALID; + write_locked(); + } + wake_.notify_all(); + if (heartbeat_.joinable()) heartbeat_.join(); + } + + bool active() const { return shared_ != nullptr; } + + void seek(uint64_t route_ns) { + if (!shared_) return; + std::lock_guard lock(lock_); + ++state_.epoch; + state_.route_ns = route_ns; + state_.host_ns = host_now(); + state_.flags = (state_.flags & PAUSED) | SEEKING; + state_.boot_offset_ns = 0; + write_locked(); + } + + void pause(bool paused) { + if (!shared_) return; + std::lock_guard lock(lock_); + reanchor_locked(host_now()); + if (paused) state_.flags |= PAUSED; + else state_.flags &= ~PAUSED; + write_locked(); + } + + void set_speed(double speed) { + if (!shared_) return; + if (!std::isfinite(speed) || speed <= 0.0) return; + std::lock_guard lock(lock_); + reanchor_locked(host_now()); + state_.speed = speed; + write_locked(); + } + + void published(uint64_t route_ns, double speed) { + if (!shared_) return; + std::lock_guard lock(lock_); + state_.route_ns = route_ns; + state_.host_ns = host_now(); + state_.speed = speed; + state_.flags = (state_.flags & (PAUSED | BOOT_KNOWN)) | VALID; + write_locked(); + } + + void observe_boot_pair(uint64_t mono_ns, uint64_t boot_ns) { + if (!shared_) return; + if (mono_ns == 0 || boot_ns < mono_ns || boot_ns - mono_ns > uint64_t(std::numeric_limits::max())) return; + std::lock_guard lock(lock_); + state_.boot_offset_ns = boot_ns - mono_ns; + state_.flags |= BOOT_KNOWN; + write_locked(); + } + +private: + void reanchor_locked(uint64_t host_ns) { + if ((state_.flags & VALID) && !(state_.flags & PAUSED) && host_ns >= state_.host_ns) { + long double route_ns = state_.route_ns + (host_ns - state_.host_ns) * static_cast(state_.speed); + if (route_ns >= std::numeric_limits::max()) state_.flags &= ~VALID; + else state_.route_ns = static_cast(route_ns); + } + state_.host_ns = host_ns; + } + + void write_locked() { + if (!shared_) return; + state_.heartbeat_ns = host_now(); + sequence_ += 2; + __atomic_store_n(&shared_->sequence, sequence_ - 1, __ATOMIC_SEQ_CST); + std::memcpy(reinterpret_cast(shared_) + sizeof(uint64_t), + reinterpret_cast(&state_) + sizeof(uint64_t), sizeof(Record) - sizeof(uint64_t)); + __atomic_store_n(&shared_->sequence, sequence_, __ATOMIC_RELEASE); + } + + int file_ = -1; + Record *shared_ = nullptr; + Record state_ = {}; + uint64_t sequence_ = 0; + bool stopped_ = false; + std::mutex lock_; + std::condition_variable wake_; + std::thread heartbeat_; +}; diff --git a/openpilot/tools/replay/display_clock.py b/openpilot/tools/replay/display_clock.py new file mode 100644 index 0000000000..dfec7b345d --- /dev/null +++ b/openpilot/tools/replay/display_clock.py @@ -0,0 +1,102 @@ +"""Read the owned replay publisher's clock without modifying recorded messages.""" + +from dataclasses import dataclass +import math +import mmap +import os +from pathlib import Path +import re +import stat +import struct +import sys +import time + + +MAGIC = 0x53505250434C4B31 +VALID, PAUSED, SEEKING, BOOT_KNOWN = 1, 2, 4, 8 +RECORD = struct.Struct("=QQIIQQQdqQQ") +HEARTBEAT_MAX_AGE_NS = 250_000_000 +_PREFIX = re.compile(r"replay-[A-Za-z0-9_-]{1,48}\Z") + + +@dataclass(frozen=True) +class ClockSample: + now_ns: int | None + boot_ns: int | None + host_ns: int + epoch: int | None = None + paused: bool = False + valid: bool = False + + +class DisplayClockReader: + @staticmethod + def enabled(env=None) -> bool: + env = os.environ if env is None else env + prefix = env.get("OPENPILOT_PREFIX", "") + root = Path("/tmp" if sys.platform == "darwin" else "/dev/shm") + path = root / ("msgq_" + prefix) / "display-clock" + return (env.get("SP_HOST_RUNTIME") == "1" and bool(_PREFIX.fullmatch(prefix)) and + env.get("SP_REPLAY_CLOCK_PATH") == str(path)) + + def __init__(self, env=None): + env = os.environ if env is None else env + self._mapping = None + prefix = env.get("OPENPILOT_PREFIX", "") + root = Path("/tmp" if sys.platform == "darwin" else "/dev/shm") + parent = root / ("msgq_" + prefix) + if not self.enabled(env): + return + directory = file = None + try: + directory = os.open(parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC) + metadata = os.fstat(directory) + if metadata.st_uid != os.getuid() or stat.S_IMODE(metadata.st_mode) != 0o700: + return + file = os.open("display-clock", os.O_RDONLY | os.O_NONBLOCK | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=directory) + metadata = os.fstat(file) + if (not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != os.getuid() or + stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_nlink != 1 or metadata.st_size != RECORD.size): + return + self._mapping = mmap.mmap(file, RECORD.size, access=mmap.ACCESS_READ) + except (OSError, ValueError): + return + finally: + if file is not None: + os.close(file) + if directory is not None: + os.close(directory) + + def close(self): + if self._mapping is not None: + self._mapping.close() + self._mapping = None + + def sample(self, host_ns: int | None = None) -> ClockSample: + host_ns = time.monotonic_ns() if host_ns is None else host_ns + missing = ClockSample(None, None, host_ns) + if self._mapping is None or type(host_ns) is not int or host_ns <= 0: + return missing + for _ in range(4): + before = struct.unpack_from("=Q", self._mapping)[0] + if not before or before % 2: + continue + values = RECORD.unpack(self._mapping[:RECORD.size]) + after = struct.unpack_from("=Q", self._mapping)[0] + if before == values[0] == after and not after % 2: + break + else: + return missing + _, magic, version, flags, epoch, route_anchor, host_anchor, speed, boot_offset, heartbeat, reserved = values + if (magic != MAGIC or version != 1 or flags & ~15 or reserved or not epoch or + not 0 < host_anchor <= heartbeat <= host_ns or host_ns - heartbeat > HEARTBEAT_MAX_AGE_NS or + not math.isfinite(speed) or speed <= 0 or flags & BOOT_KNOWN and boot_offset < 0): + return missing + paused = bool(flags & PAUSED) + if not flags & VALID or flags & SEEKING or not route_anchor: + return ClockSample(None, None, host_ns, epoch, paused) + now_ns = route_anchor if paused else route_anchor + int((host_ns - host_anchor) * speed) + boot_ns = now_ns + boot_offset if flags & BOOT_KNOWN else None + if not 0 < now_ns < 2**64 or boot_ns is not None and not 0 < boot_ns < 2**64: + return missing + return ClockSample(now_ns, boot_ns, host_ns, epoch, paused, True) diff --git a/openpilot/tools/replay/onroad.py b/openpilot/tools/replay/onroad.py index 3f942b6b94..e375ade67f 100644 --- a/openpilot/tools/replay/onroad.py +++ b/openpilot/tools/replay/onroad.py @@ -15,12 +15,13 @@ import sys import tempfile import time -USAGE = """Usage: ./onroad [jobs] [--c3|--c4|--all|--replay-only] [--alert] [--cem] [--csc] [--prefix replay-NAME] +USAGE = """Usage: ./onroad [jobs] [--c3|--c4|--all|--replay-only] [--params /saved/params] [--alert] [--cem] [--csc] [--prefix replay-NAME] Replay arguments include a route or --demo, --start, --cache, --playback, --data_dir and --no-loop. The native replay terminal provides playback controls. Route playback may read/download the route you request. --alert previews a synthetic visual critical alert; it does not publish car control. --cem and --csc preview synthetic onroad visuals in the selected native UI only. Old --nav, --offroad and --galaxy demos are not available in this port. +--params supplements recorded settings with a read-only snapshot, including unlogged layout preferences. """ UNAVAILABLE = frozenset(("--nav", "-nav", "--offroad", "--galaxy")) PREFIX_RE = re.compile(r"replay-[A-Za-z0-9_-]{1,48}\Z") @@ -35,6 +36,7 @@ class OnroadPlan: replay_args: tuple[str, ...] alert: bool = False visual_preview: frozenset[str] = frozenset() + params_snapshot: str | None = None def _block_alert_service(args: tuple[str, ...]) -> list[str]: @@ -65,6 +67,7 @@ def parse_onroad_args(args: list[str]) -> OnroadPlan: alert = False visual_preview: set[str] = set() prefix = None + params_snapshot = None replay_args: list[str] = [] index = 0 while index < len(args): @@ -95,6 +98,16 @@ def parse_onroad_args(args: list[str]) -> OnroadPlan: else: targets.append(arg.removeprefix("--")) explicit = True + elif arg == "--params" or arg.startswith("--params="): + if arg == "--params": + if index + 1 >= len(args): + raise ValueError("missing --params directory") + params_snapshot = args[index + 1] + index += 1 + else: + params_snapshot = arg.partition("=")[2] + if not params_snapshot or not Path(params_snapshot).is_absolute(): + raise ValueError("--params must name an absolute directory") elif arg in ("--prefix", "-p"): if index + 1 >= len(args): raise ValueError("missing --prefix value") @@ -136,7 +149,7 @@ def parse_onroad_args(args: list[str]) -> OnroadPlan: if (alert or visual_preview) and (replay_only or (explicit and not targets)): raise ValueError("Visual previews require a native UI") return OnroadPlan(tuple(name for name in ("c3", "c4") if name in targets), explicit, replay_only, - prefix, tuple(replay_args), alert, frozenset(visual_preview)) + prefix, tuple(replay_args), alert, frozenset(visual_preview), params_snapshot) def _private_prefix(requested: str | None) -> str: @@ -178,11 +191,11 @@ def _parent_prefix(prefix: str): os.environ["OPENPILOT_PREFIX"] = previous -def seed_replay_params(init_data, params_root: str, prefix: str) -> int: +def seed_replay_params(init_data, params_root: str, prefix: str, snapshot: str | None = None) -> int: from openpilot.common.params import Params from openpilot.tools.replay.onroad_config import seed_preview with _parent_prefix(prefix): - return seed_preview(init_data, Params(params_root)) + return seed_preview(init_data, Params(params_root), snapshot) def _stop(children: list[subprocess.Popen]) -> None: @@ -240,7 +253,7 @@ def supervise(replay_command: list[str], ui_commands: list[list[str]], env: dict def run(plan: OnroadPlan) -> int: - from openpilot.tools.replay.onroad_config import parse_replay_args, route_init_data, select_ui_target + from openpilot.tools.replay.onroad_config import parse_replay_args, route_init_data, select_ui_target, replay_device_type from openpilot.starpilot.ui.host_launch import launch_environment from openpilot.starpilot.ui.developer_preview import encode_flags from openpilot.starpilot.ui.presentation import Profile @@ -278,15 +291,27 @@ def run(plan: OnroadPlan) -> int: host_params.parent.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix="replay-params-", dir=host_params.parent) as params_root: env = {**source, "OPENPILOT_PREFIX": prefix, "PARAMS_ROOT": params_root, - "SP_ONROAD_VISUAL_PREVIEW": preview} + "SP_ONROAD_VISUAL_PREVIEW": preview, "SP_REPLAY_DEVICE_TYPE": replay_device_type(init_data)} env["COMMA_CACHE"] = str(host_params.parent / "download-cache") - seed_replay_params(init_data, params_root, prefix) + restored = seed_replay_params(init_data, params_root, prefix, plan.params_snapshot) marker.mkdir(mode=0o700) try: ui_commands = [[sys.executable, "-m", "openpilot.starpilot.ui.host_launch", "large" if target == "c3" else "compact"] for target in targets] print(f"Replay prefix {prefix}; native UI: {', '.join(targets) if targets else 'none'}", flush=True) - with owned_ipc_namespace(prefix): + if targets: + print(f"Recorded device: {env['SP_REPLAY_DEVICE_TYPE']}; restored {restored} saved settings", flush=True) + if plan.params_snapshot is None: + print("Unlogged settings use defaults; --params /saved/params supplies an exact saved layout and preferences.", flush=True) + with owned_ipc_namespace(prefix) as ipc_root: + from openpilot.tools.replay.display_clock import RECORD + clock_path = ipc_root / "display-clock" + fd = os.open(clock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + try: + os.ftruncate(fd, RECORD.size) + finally: + os.close(fd) + env["SP_REPLAY_CLOCK_PATH"] = str(clock_path) replay_command = [str(replay_binary), *(_block_alert_service(plan.replay_args) if plan.alert else plan.replay_args)] demos = [[sys.executable, "-m", "openpilot.tools.replay.alert_demo"]] if plan.alert else [] return supervise(replay_command, ui_commands, env, demo_commands=demos) diff --git a/openpilot/tools/replay/onroad_config.py b/openpilot/tools/replay/onroad_config.py index ae3d0fbacb..10adaa2719 100644 --- a/openpilot/tools/replay/onroad_config.py +++ b/openpilot/tools/replay/onroad_config.py @@ -1,18 +1,26 @@ -"""Route metadata and bounded saved-display seeding for disposable host replay.""" +"""Route metadata and recorded settings for disposable host replay.""" from dataclasses import dataclass from pathlib import Path from collections.abc import Sequence +import json +import math +import os +import stat import sys -from openpilot.common.params import Params +from openpilot.common.params import Params, ParamKeyType, UnknownKeyName from openpilot.common.version import terms_version, training_version +from openpilot.starpilot.schema_cache import CACHE_KEYS, inspect_cache from openpilot.tools.lib.logreader import LogReader, ReadMode, parse_direct, parse_indirect from openpilot.tools.lib.route import SegmentRange DEMO_ROUTE = "5beb9b58bd12b691/0000010a--a51155e496" -DISPLAY_BOOL_KEYS = frozenset(("IsMetric", "HideSpeed", "HideMaxSpeed", "HideSteeringWheel", "EnableTorqueBarWidget")) +PRIVATE_KEYS = frozenset(("AccessToken", "GithubSshKeys", "GithubUsername", "SecOCKey", "PairingEmail", + "AthenadUploadQueue", "AthenadRecentlyViewedRoutes", "UsageStatsState", "UsageStatsStatus")) +MAX_PARAM_BYTES = 32 * 1024 * 1024 +DEVICE_TYPES = frozenset(("tici", "tizi", "mici")) VALUE_OPTIONS = frozenset(("-a", "--allow", "-b", "--block", "-c", "--cache", "-s", "--start", "-x", "--playback", "-d", "--data_dir")) FLAG_OPTIONS = frozenset(("--cabin", "--dcam", "--wide-road", "--ecam", "--no-loop", "--no-cache", @@ -103,8 +111,38 @@ def select_ui_target(init_data) -> str: return "c4" if str(getattr(init_data, "deviceType", "")).lower() in ("mici", "c4") else "c3" +def replay_device_type(init_data) -> str: + value = str(getattr(init_data, "deviceType", "")).lower() + return value if value in DEVICE_TYPES else "pc" + + +def seed_param(params: Params, key: str, raw: bytes) -> bool: + if key in PRIVATE_KEYS or not raw or len(raw) > MAX_PARAM_BYTES: + return False + try: + kind = params.get_type(key) + if kind == ParamKeyType.BYTES: + if key not in CACHE_KEYS or inspect_cache(key, raw).status != "valid": + return False + value = raw + elif kind == ParamKeyType.BOOL: + if raw not in (b"0", b"1"): + return False + value = raw == b"1" + else: + value = params.cpp2python(key, raw) + if value is None or kind == ParamKeyType.FLOAT and not math.isfinite(value): + return False + if kind == ParamKeyType.JSON: + json.dumps(value, allow_nan=False) + params.put(key, value, block=True) + return True + except (UnknownKeyName, UnicodeError, TypeError, ValueError): + return False + + def seed_display_params(init_data, params: Params) -> int: - """Copy only exact, registered display booleans; no tokens or opaque docs.""" + """Restore registered typed settings and verified cache envelopes, never credentials.""" try: entries = init_data.params.entries except (AttributeError, TypeError): @@ -112,19 +150,40 @@ def seed_display_params(init_data, params: Params) -> int: count = 0 for entry in entries: name = str(entry.key) - if name not in DISPLAY_BOOL_KEYS: - continue raw = bytes(entry.value) - if raw not in (b"0", b"1"): + count += seed_param(params, name, raw) + return count + + +def seed_snapshot(directory: str, params: Params) -> int: + root = Path(directory) + if not root.is_absolute() or not root.is_dir(): + raise ValueError("--params must name an absolute directory of saved Param files") + count = 0 + for key in params.all_keys(): + name = key.decode() + if name in PRIVATE_KEYS: continue - params.put_bool(name, raw == b"1", block=True) - count += 1 + try: + fd = os.open(root / name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + except FileNotFoundError: + continue + try: + info = os.fstat(fd) + if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_PARAM_BYTES: + raise ValueError(f"Invalid saved Param file: {name}") + with os.fdopen(fd, "rb", closefd=False) as file: + count += seed_param(params, name, file.read(MAX_PARAM_BYTES + 1)) + finally: + os.close(fd) return count -def seed_preview(init_data, params: Params) -> int: +def seed_preview(init_data, params: Params, snapshot: str | None = None) -> int: count = seed_display_params(init_data, params) - params.put("HasAcceptedTerms", terms_version, block=True) - params.put("CompletedTrainingVersion", training_version, block=True) - params.put_bool("OpenpilotEnabledToggle", True, block=True) + if snapshot is not None: + count += seed_snapshot(snapshot, params) + for key, value in (("HasAcceptedTerms", terms_version), ("CompletedTrainingVersion", training_version)): + if params.get(key) is None: + params.put(key, value, block=True) return count diff --git a/openpilot/tools/replay/replay.cc b/openpilot/tools/replay/replay.cc index 9a7b5ec62c..04a384e534 100644 --- a/openpilot/tools/replay/replay.cc +++ b/openpilot/tools/replay/replay.cc @@ -90,6 +90,7 @@ void Replay::stop() { stream_thread_.join(); rInfo("shutdown: done"); } + display_clock_.stop(); } bool Replay::load() { @@ -134,6 +135,7 @@ void Replay::seekTo(double seconds, bool relative) { interruptStream([&]() { current_segment_.store(target_segment); cur_mono_time_ = route_start_ts_ + target_time * 1e9; + display_clock_.seek(cur_mono_time_); cur_which_ = cereal::Event::Which::INIT_DATA; seeking_to_.store(target_time, std::memory_order_relaxed); return false; @@ -166,11 +168,21 @@ void Replay::pause(bool pause) { interruptStream([=]() { rWarning("%s at %.2f s", pause ? "paused..." : "resuming", currentSeconds()); user_paused_ = pause; + display_clock_.pause(pause); return !pause; }); } } +void Replay::setSpeed(float speed) { + if (!std::isfinite(speed) || speed <= 0.0f) return; + interruptStream([&]() { + speed_ = speed; + display_clock_.set_speed(speed); + return events_ready_; + }); +} + void Replay::handleSegmentMerge() { if (exit_) return; @@ -244,8 +256,8 @@ void Replay::startStream(const std::shared_ptr segment) { stream_thread_ = std::thread(&Replay::streamThread, this); } -void Replay::publishMessage(const Event *e) { - if (event_filter_ && event_filter_(e)) return; +bool Replay::publishMessage(const Event *e) { + if (event_filter_ && event_filter_(e)) return false; if (!sm_) { if (!pm_) pm_ = std::make_unique(active_services_); // consumers with an event filter never need one @@ -254,12 +266,14 @@ void Replay::publishMessage(const Event *e) { if (ret == -1) { rWarning("stop publishing %s due to multiple publishers error", sockets_[e->which]); sockets_[e->which] = nullptr; + return false; } } else { capnp::FlatArrayMessageReader reader(e->data); auto event = reader.getRoot(); sm_->update_msgs(nanos_since_boot(), {{sockets_[e->which], event}}); } + return true; } void Replay::publishFrame(const Event *e) { @@ -400,7 +414,29 @@ std::vector::const_iterator Replay::publishEvents(std::vector::con if (interrupt_requested_) break; if (evt.eidx_segnum == -1) { - publishMessage(&evt); + bool published = publishMessage(&evt); + if (published && display_clock_.active()) { + // These Python publishers use CLOCK_MONOTONIC; native envelopes can use CLOCK_BOOTTIME. + if (evt.which == cereal::Event::Which::SELFDRIVE_STATE || evt.which == cereal::Event::Which::CAR_CONTROL || + evt.which == cereal::Event::Which::MODEL_V2) { + display_clock_.published(evt.mono_time, speed_); + } + if (evt.which == cereal::Event::Which::SPOT_MONITOR_STATE) { + capnp::FlatArrayMessageReader reader(evt.data); + auto event = reader.getRoot(); + auto pair = event.getSpotMonitorState().getObservation(); + const auto mono = pair.getObservedMonoTime(); + const auto boot = pair.getObservedBootTime(); + const auto eof = pair.getSourceFrameEofBootTime(); + if (event.getValid() && pair.getVersion() == 1 && pair.getSequence() > 0 && + mono > 0 && mono == evt.mono_time && eof > 0 && eof <= boot && boot - eof <= 500000000ULL && + eof <= std::numeric_limits::max() - 500000000ULL && + pair.getValidUntilBootTime() == eof + 500000000ULL) { + display_clock_.published(mono, speed_); + display_clock_.observe_boot_pair(mono, boot); + } + } + } } else if (camera_server_) { if (speed_ > 1.0) { camera_server_->waitForSent(); diff --git a/openpilot/tools/replay/replay.h b/openpilot/tools/replay/replay.h index 59e1d67d48..ec0aa92144 100644 --- a/openpilot/tools/replay/replay.h +++ b/openpilot/tools/replay/replay.h @@ -9,6 +9,7 @@ #include #include "tools/replay/camera.h" +#include "tools/replay/display_clock.h" #include "tools/replay/seg_mgr.h" #include "tools/replay/timeline.h" @@ -57,7 +58,7 @@ public: inline double toSeconds(uint64_t mono_time) const { return (mono_time - route_start_ts_) / 1e9; } inline double minSeconds() const { return min_seconds_; } inline double maxSeconds() const { return max_seconds_; } - inline void setSpeed(float speed) { speed_ = speed; } + void setSpeed(float speed); inline float getSpeed() const { return speed_; } inline const std::string &carFingerprint() const { return car_fingerprint_; } inline const std::shared_ptr> getTimeline() const { return timeline_.getEntries(); } @@ -84,12 +85,13 @@ private: std::vector::const_iterator last, int &last_processed_segment, uint64_t &segment_start_time); - void publishMessage(const Event *e); + bool publishMessage(const Event *e); void publishFrame(const Event *e); void checkSeekProgress(); std::unique_ptr seg_mgr_; Timeline timeline_; + ReplayDisplayClock display_clock_; pthread_t stream_thread_id = 0; std::thread stream_thread_; diff --git a/openpilot/tools/replay/tests/test_onroad_host.py b/openpilot/tools/replay/tests/test_onroad_host.py index ff527929cd..7e14150530 100644 --- a/openpilot/tools/replay/tests/test_onroad_host.py +++ b/openpilot/tools/replay/tests/test_onroad_host.py @@ -5,6 +5,7 @@ from types import SimpleNamespace import os import select import signal +import shlex import subprocess import sys import tempfile @@ -14,7 +15,7 @@ from unittest.mock import patch from openpilot.common.params import Params from openpilot.tools.replay.onroad import _block_alert_service, _ipc_root, owned_ipc_namespace, parse_onroad_args, run, seed_replay_params, supervise -from openpilot.tools.replay.onroad_config import first_log_identifier, parse_replay_args, seed_preview, select_ui_target +from openpilot.tools.replay.onroad_config import first_log_identifier, parse_replay_args, seed_preview, select_ui_target, replay_device_type class TestOnroadHost(unittest.TestCase): @@ -243,19 +244,76 @@ assert seen[-2:] == ['full', 'clear'], seen local.unlink() self.assertIsNone(first_log_identifier(args)) - def test_seed_only_registered_display_booleans_in_disposable_params(self): + def test_seed_recorded_typed_settings_and_preserve_explicit_choices(self): with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, OPENPILOT_PREFIX="replay-test-seed"): params = Params(directory) entries = [SimpleNamespace(key="IsMetric", value=b"1"), SimpleNamespace(key="HideSpeed", value=b"0"), SimpleNamespace(key="HideMaxSpeed", value=b"not-a-bool"), + SimpleNamespace(key="CameraView", value=b"3"), + SimpleNamespace(key="OpenpilotEnabledToggle", value=b"0"), + SimpleNamespace(key="ConditionalModeConfig", value=b'{"version":1,"mode":"cem"}'), + SimpleNamespace(key="LaneCenteringStrength", value=b"nan"), + SimpleNamespace(key="CarParamsPersistent", value=b"unversioned"), SimpleNamespace(key="AccessToken", value=b"secret")] init = SimpleNamespace(params=SimpleNamespace(entries=entries)) - self.assertEqual(seed_preview(init, params), 2) + self.assertEqual(seed_preview(init, params), 5) self.assertIs(params.get("IsMetric"), True) self.assertIs(params.get("HideSpeed"), False) self.assertIsNone(params.get("HideMaxSpeed")) self.assertIsNone(params.get("AccessToken")) - self.assertIs(params.get("OpenpilotEnabledToggle"), True) + self.assertIsNone(params.get("CarParamsPersistent")) + self.assertIsNone(params.get("LaneCenteringStrength")) + self.assertEqual(params.get("CameraView"), 3) + self.assertEqual(params.get("ConditionalModeConfig"), {"version": 1, "mode": "cem"}) + self.assertIs(params.get("OpenpilotEnabledToggle"), False) + + def test_saved_snapshot_supplements_unlogged_preferences_without_modifying_source(self): + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, OPENPILOT_PREFIX="replay-test-snapshot"): + root = Path(directory) + saved = root / "saved" + saved.mkdir() + raw = b'{"version":1,"mode":"stock"}' + (saved / "ConditionalModeConfig").write_bytes(raw) + (saved / "HideSpeed").write_bytes(b"1") + (saved / "AccessToken").write_bytes(b"secret") + params = Params(str(root / "target")) + init = SimpleNamespace(params=SimpleNamespace(entries=[SimpleNamespace(key="HideSpeed", value=b"0")])) + self.assertEqual(seed_preview(init, params, str(saved)), 3) + self.assertIs(params.get("HideSpeed"), True) + self.assertEqual(params.get("ConditionalModeConfig"), {"version": 1, "mode": "stock"}) + self.assertIsNone(params.get("AccessToken")) + self.assertEqual((saved / "ConditionalModeConfig").read_bytes(), raw) + plan = parse_onroad_args(["--params", str(saved), "--demo"]) + self.assertEqual(plan.params_snapshot, str(saved)) + self.assertNotIn("--params", plan.replay_args) + + def test_saved_snapshot_rejects_symlinks_and_replay_device_keeps_tizi_identity(self): + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, OPENPILOT_PREFIX="replay-test-snapshot"): + root = Path(directory) + saved = root / "saved" + saved.mkdir() + (saved / "IsMetric").symlink_to(root / "missing") + with self.assertRaises(OSError): + seed_preview(None, Params(str(root / "target")), str(saved)) + self.assertEqual(replay_device_type(SimpleNamespace(deviceType="tizi")), "tizi") + self.assertEqual(replay_device_type(SimpleNamespace(deviceType="mici")), "mici") + self.assertEqual(replay_device_type(None), "pc") + + def test_seed_valid_cache_retains_its_existing_schema_provenance(self): + from opendbc.car import structs as car + from openpilot.starpilot.schema_cache import get_cache, put_cache + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, OPENPILOT_PREFIX="replay-test-cache"): + source = Params(str(Path(directory) / "source")) + cp = car.CarParams.new_message(carFingerprint="HYUNDAI IONIQ 6 2023", openpilotLongitudinalControl=True) + put_cache(source, "CarParamsPersistent", cp, block=True) + raw = Path(source.get_param_path("CarParamsPersistent")).read_bytes() + target = Params(str(Path(directory) / "target")) + init = SimpleNamespace(params=SimpleNamespace(entries=[SimpleNamespace(key="CarParamsPersistent", value=raw)])) + self.assertEqual(seed_preview(init, target), 1) + with car.CarParams.from_bytes(get_cache(target, "CarParamsPersistent")) as restored: + self.assertEqual(restored.carFingerprint, cp.carFingerprint) + self.assertTrue(restored.openpilotLongitudinalControl) + self.assertEqual(Path(target.get_param_path("CarParamsPersistent")).read_bytes(), raw) def test_seed_uses_replay_namespace_not_runner_namespace(self): with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, OPENPILOT_PREFIX="starpilot-dev-original"): @@ -320,3 +378,203 @@ assert seen[-2:] == ['full', 'clear'], seen if proc.poll() is None: proc.kill() proc.wait(timeout=2) + + +class TestReplayDisplayClock(unittest.TestCase): + def test_epoch_reset_clears_real_submaster_structs_and_lists(self): + from openpilot.cereal import messaging + from openpilot.selfdrive.ui.ui_state import UIState + state = object.__new__(UIState) + with patch.dict(os.environ, OPENPILOT_PREFIX=self.prefix): + state.sm = messaging.SubMaster(['carControl', 'modelV2', 'pandaStates', 'onroadEvents']) + state.sm.update_msgs(time.monotonic(), [messaging.new_message('carControl', valid=True), + messaging.new_message('modelV2', valid=True), + messaging.new_message('pandaStates', 1, valid=True), + messaging.new_message('onroadEvents', 1, valid=True)]) + state.started = True + state.chestnut_output_seen = True + state._reset_replay_state() + self.assertFalse(state.started or state.chestnut_output_seen) + self.assertEqual(len(state.sm['pandaStates']), 0) + self.assertEqual(len(state.sm['onroadEvents']), 0) + self.assertFalse(any(state.sm.valid.values())) + self.assertFalse(any(state.sm.seen.values())) + self.assertFalse(state.sm['carControl'].latActive) + + def test_native_writer_and_python_reader_share_host_clock_and_lifecycle(self): + source = '''#include +#include "tools/replay/display_clock.h" +int main() { + ReplayDisplayClock clock; + if (!clock.active()) return 2; + clock.published(30000000000ULL, 1.0); + std::cout << "ready" << std::endl; + char command; + while (std::cin >> command) { + if (command == 'p') clock.pause(true); + if (command == 's') clock.seek(2000000000ULL); + if (command == 'n') clock.published(2000000000ULL, 1.0); + if (command == 'q') { clock.stop(); break; } + std::cout << "done" << std::endl; + } +}''' + with tempfile.TemporaryDirectory() as directory: + cpp, binary = Path(directory) / "clock.cc", Path(directory) / "clock" + cpp.write_text(source) + include = Path(__file__).resolve().parents[3] + compiler = shlex.split(os.environ.get('CXX', 'c++')) + subprocess.run([*compiler, '-std=c++17', '-pthread', '-I', str(include), str(cpp), '-o', str(binary)], + check=True, capture_output=True, timeout=30) + process = subprocess.Popen([str(binary)], env={**os.environ, **self.env}, stdin=subprocess.PIPE, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + try: + self.assertEqual(process.stdout.readline().strip(), 'ready') + sample = self.reader.sample() + self.assertTrue(sample.valid) + self.assertLess(abs(sample.now_ns - 30_000_000_000), 250_000_000) + process.stdin.write('p\n') + process.stdin.flush() + self.assertEqual(process.stdout.readline().strip(), 'done') + paused = self.reader.sample() + time.sleep(.3) + self.assertTrue(self.reader.sample().valid) + self.assertEqual(self.reader.sample().now_ns, paused.now_ns) + process.stdin.write('s\n') + process.stdin.flush() + self.assertEqual(process.stdout.readline().strip(), 'done') + seek = self.reader.sample() + self.assertFalse(seek.valid) + self.assertEqual(seek.epoch, sample.epoch + 1) + process.stdin.write('n\n') + process.stdin.flush() + self.assertEqual(process.stdout.readline().strip(), 'done') + self.assertEqual(self.reader.sample().now_ns, 2_000_000_000) + process.stdin.write('q\n') + process.stdin.flush() + self.assertEqual(process.wait(timeout=3), 0) + self.assertFalse(self.reader.sample().valid) + finally: + if process.poll() is None: + process.kill() + process.wait(timeout=3) + + def setUp(self): + from openpilot.tools.replay.display_clock import DisplayClockReader, RECORD + self.reader_type = DisplayClockReader + self.record = RECORD + self.prefix = f"replay-clocktest-{os.getpid()}-{time.time_ns()}" + self.root = self.enterContext(owned_ipc_namespace(self.prefix)) + self.path = self.root / "display-clock" + self.path.write_bytes(bytes(RECORD.size)) + self.path.chmod(0o600) + self.env = {"SP_HOST_RUNTIME": "1", "OPENPILOT_PREFIX": self.prefix, "SP_REPLAY_CLOCK_PATH": str(self.path)} + self.reader = DisplayClockReader(self.env) + self.addCleanup(self.reader.close) + + def write(self, **changes): + from openpilot.tools.replay.display_clock import MAGIC, VALID + fields = {"sequence": 2, "magic": MAGIC, "version": 1, "flags": VALID, "epoch": 1, + "route_ns": 30_000_000_000, "host_ns": 10_000_000_000, "speed": 1., "boot_offset": 0, + "heartbeat": 10_000_000_000, "reserved": 0} + fields.update(changes) + self.path.write_bytes(self.record.pack(*fields.values())) + + def test_recorded_time_speed_and_explicit_pause_keep_clock_domains_separate(self): + from openpilot.tools.replay.display_clock import PAUSED, VALID + self.write(speed=2.) + sample = self.reader.sample(10_100_000_000) + self.assertTrue(sample.valid) + self.assertEqual(sample.now_ns, 30_200_000_000) + self.assertEqual(sample.host_ns, 10_100_000_000) + self.assertIsNone(sample.boot_ns) + self.write(flags=VALID | PAUSED, route_ns=30_200_000_000, host_ns=10_100_000_000, + heartbeat=10_400_000_000) + sample = self.reader.sample(10_500_000_000) + self.assertTrue(sample.valid and sample.paused) + self.assertEqual(sample.now_ns, 30_200_000_000) + self.write(route_ns=30_200_000_000, host_ns=10_500_000_000, heartbeat=10_500_000_000, speed=.5) + self.assertEqual(self.reader.sample(10_600_000_000).now_ns, 30_250_000_000) + + def test_unknown_boot_offset_differs_from_valid_zero_and_recorded_suspend_offset(self): + from openpilot.tools.replay.display_clock import BOOT_KNOWN, VALID + self.write() + self.assertIsNone(self.reader.sample(10_000_000_000).boot_ns) + self.write(flags=VALID | BOOT_KNOWN) + self.assertEqual(self.reader.sample(10_000_000_000).boot_ns, 30_000_000_000) + self.write(flags=VALID | BOOT_KNOWN, boot_offset=9_000_000_000) + self.assertEqual(self.reader.sample(10_000_000_000).boot_ns, 39_000_000_000) + + def test_seek_epoch_is_visible_but_invalid_until_new_publication(self): + from openpilot.tools.replay.display_clock import SEEKING, VALID + self.write() + self.assertEqual(self.reader.sample(10_000_000_000).epoch, 1) + self.write(flags=SEEKING, epoch=2, route_ns=2_000_000_000) + sample = self.reader.sample(10_000_000_000) + self.assertFalse(sample.valid) + self.assertEqual(sample.epoch, 2) + self.assertIsNone(sample.now_ns) + self.write(flags=VALID, epoch=2, route_ns=2_000_000_000) + sample = self.reader.sample(10_100_000_000) + self.assertTrue(sample.valid) + self.assertEqual((sample.epoch, sample.now_ns), (2, 2_100_000_000)) + + def test_dead_publisher_torn_record_and_invalid_metadata_fail_closed(self): + from openpilot.tools.replay.display_clock import BOOT_KNOWN, HEARTBEAT_MAX_AGE_NS, VALID + self.write() + self.assertTrue(self.reader.sample(10_000_000_000 + HEARTBEAT_MAX_AGE_NS).valid) + self.assertFalse(self.reader.sample(10_000_000_001 + HEARTBEAT_MAX_AGE_NS).valid) + for changes in ({"sequence": 3}, {"sequence": 0}, {"magic": 0}, {"version": 2}, {"flags": 16}, + {"epoch": 0}, {"speed": float("nan")}, {"speed": 0}, {"reserved": 1}, + {"heartbeat": 10_000_000_001}, {"host_ns": 10_000_000_001}, + {"flags": VALID | BOOT_KNOWN, "boot_offset": -1}): + with self.subTest(changes=changes): + self.write(**changes) + self.assertFalse(self.reader.sample(10_000_000_000).valid) + + def test_only_exact_owned_context_path_regular_file_size_and_modes_are_admitted(self): + self.reader.close() + self.write() + for changes in ({"SP_HOST_RUNTIME": "0"}, {"OPENPILOT_PREFIX": "replay-other"}, + {"SP_REPLAY_CLOCK_PATH": str(self.root / "other")}, {"OPENPILOT_PREFIX": "../replay-other"}): + with self.subTest(changes=changes): + self.assertFalse(self.reader_type.enabled({**self.env, **changes})) + reader = self.reader_type({**self.env, **changes}) + self.assertFalse(reader.sample(10_000_000_000).valid) + reader.close() + self.path.chmod(0o644) + reader = self.reader_type(self.env) + self.assertFalse(reader.sample(10_000_000_000).valid) + reader.close() + + self.path.chmod(0o600) + for size in (self.record.size - 1, self.record.size + 1): + self.path.write_bytes(bytes(size)) + reader = self.reader_type(self.env) + self.assertFalse(reader.sample(10_000_000_000).valid) + reader.close() + self.write() + self.root.chmod(0o750) + reader = self.reader_type(self.env) + self.assertFalse(reader.sample(10_000_000_000).valid) + reader.close() + self.root.chmod(0o700) + target = self.root / "other" + self.path.rename(target) + self.path.symlink_to(target) + reader = self.reader_type(self.env) + self.assertFalse(reader.sample(10_000_000_000).valid) + reader.close() + self.path.unlink() + target.rename(self.path) + os.link(self.path, target) + reader = self.reader_type(self.env) + self.assertFalse(reader.sample(10_000_000_000).valid) + reader.close() + + def test_enabled_context_stays_enabled_when_clock_file_is_missing(self): + self.reader.close() + self.path.unlink() + self.assertTrue(self.reader_type.enabled(self.env)) + reader = self.reader_type(self.env) + self.assertFalse(reader.sample(10_000_000_000).valid) + reader.close() diff --git a/tools/STARPILOT_DEVELOPMENT.md b/tools/STARPILOT_DEVELOPMENT.md index 841290687e..20610e3127 100644 --- a/tools/STARPILOT_DEVELOPMENT.md +++ b/tools/STARPILOT_DEVELOPMENT.md @@ -108,6 +108,22 @@ a private session name. Playback uses the current native terminal controls. Route access may require authentication or download the route the developer requests. This is log playback; it does not run a driving manager or send CAN. +The UI uses the route's recorded device type, registered saved settings and +compatible versioned caches. Message freshness follows recorded time, including +playback speed, pause and seeking; recorded messages are not rewritten. Seeking +clears the previous display observations. A dead replay publisher invalidates +the display clock even while paused. Recorded wheel actions are not executed. +An explicit `--c3`, `--c4` or `--all` overrides the screen format while retaining +the recorded device identity. + +Some preferences, including custom layouts, may be absent from the log. Supply +`--params /absolute/path/to/saved/params` to supplement them with a flat saved +Params snapshot. The snapshot is read only; credentials and unqualified binary +caches are excluded. Without that snapshot, missing settings use defaults, so +the preview cannot reproduce unlogged preferences. Displays requiring both +recorded monotonic and boot clocks remain unavailable until a recorded clock +pair establishes their relationship. + `./onroad -alert` previews a synthetic critical visual alert in the private replay session; it never publishes car control. `./onroad --cem` and `--csc` show labeled synthetic CEM and curve visuals in that private UI session. They