Recorded replay state

This commit is contained in:
firestar5683
2026-10-07 22:06:02 -05:00
parent 3fa904765c
commit ce09aec43d
16 changed files with 911 additions and 47 deletions
+7
View File
@@ -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):
+4 -3
View File
@@ -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()
+56 -5
View File
@@ -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()
+4
View File
@@ -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:
+47 -9
View File
@@ -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,
@@ -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")
@@ -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):
+3
View File
@@ -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
+194
View File
@@ -0,0 +1,194 @@
#pragma once
#include <chrono>
#include <cmath>
#include <condition_variable>
#include <cstdint>
#include <cstdlib>
#include <cstring>
#include <fcntl.h>
#include <limits>
#include <mutex>
#include <string>
#include <sys/file.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <thread>
#include <unistd.h>
#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<Record *>(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<int64_t>::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<long double>(state_.speed);
if (route_ns >= std::numeric_limits<uint64_t>::max()) state_.flags &= ~VALID;
else state_.route_ns = static_cast<uint64_t>(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<char *>(shared_) + sizeof(uint64_t),
reinterpret_cast<const char *>(&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_;
};
+102
View File
@@ -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)
+33 -8
View File
@@ -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] <replay args>
USAGE = """Usage: ./onroad [jobs] [--c3|--c4|--all|--replay-only] [--params /saved/params] [--alert] [--cem] [--csc] [--prefix replay-NAME] <replay args>
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)
+72 -13
View File
@@ -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
+39 -3
View File
@@ -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> 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<PubMaster>(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<cereal::Event>();
sm_->update_msgs(nanos_since_boot(), {{sockets_[e->which], event}});
}
return true;
}
void Replay::publishFrame(const Event *e) {
@@ -400,7 +414,29 @@ std::vector<Event>::const_iterator Replay::publishEvents(std::vector<Event>::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<cereal::Event>();
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<uint64_t>::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();
+4 -2
View File
@@ -9,6 +9,7 @@
#include <vector>
#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<std::vector<Timeline::Entry>> getTimeline() const { return timeline_.getEntries(); }
@@ -84,12 +85,13 @@ private:
std::vector<Event>::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<SegmentManager> seg_mgr_;
Timeline timeline_;
ReplayDisplayClock display_clock_;
pthread_t stream_thread_id = 0;
std::thread stream_thread_;
@@ -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 <iostream>
#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()
+16
View File
@@ -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