mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-10-27 00:23:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add exclusive parked SpacePilot mode and native Big UI

Browse Source 
Use the supported Offroad lifecycle with fresh Park, receive-only CAN,
stopped-process and noOutput checks. Add offroad touch/controller play
and the native Big UI five-version-tap entry. Exiting never restores
driving mode automatically within the existing manager session.

Qualified by210 unique focused passing cases, lint/type checks, native
host rendering/input/resource checks and independent lifecycle review.
Actual vehicle lifecycle, hardware inputs and device headroom remain
unqualified; no deployment is part of this change.
This commit is contained in:
firestar5683
2026-10-26 00:01:36 -05:00
parent 35f62c3c5a
commit 347397ad38
17 changed files with 1245 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
openpilot/selfdrive/ui/ui.py
+5 -1
View File
@@ -44,6 +44,10 @@ def main():
from openpilot.starpilot.ui.layout_preview_runtime import LayoutPreviewRuntime
layout = StarMainLayout() if BIG_UI else StarMiciMainLayout()
from openpilot.starpilot.controllers.runtime import ControllerRuntime
from openpilot.starpilot.ui.spacepilot.native import SpacePilotPage
def game_input():
page = gui_app.get_active_widget()
return page.capture_controller if isinstance(page, SpacePilotPage) else None
from openpilot.starpilot.galaxy.settings import LiveContextSource
# Both runtimes read the UI collector; neither advances or closes its sockets.
controller_authority = LiveContextSource(ui_state.params, messages=ui_state.sm,
@@ -52,7 +56,7 @@ def main():
borrowed_messages=True, evidence_wait_ms=0)
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)
invoke_favorite=layout.star.favorites_owner.invoke, authority=controller_authority, input_capture=game_input)
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:
openpilot/starpilot/controllers/owner.py
+13 -1
View File
@@ -239,7 +239,7 @@ class ControllerOwner:
with self._lock:
self._last_press = {"deviceId": device_id, "code": code, "slot": slot, "executed": executed, "message": message}
def feed(self, press):
def feed(self, press, *, capture: Callable[[str | None, int], None] | None = None):
device_id, code = _field(press, "device_id"), _field(press, "code")
timestamp = _field(press, "timestamp_ns")
now = self.clock()
@@ -255,6 +255,18 @@ class ControllerOwner:
self._sequences = {key: seen for key, seen in self._sequences.items() if seen >= now - PRESS_AGE_NS}
learning = self._learning
testing = bool(self._testing_until)
if capture is not None:
# Native modal input owns this whole batch. Resolve the saved logical key
# without invoking its vehicle action or modifying a learn/test session.
document, _, readable, valid = _read_document(self.params)
key = None
if readable and valid and document["enabled"]:
binding = next((item for item in document["bindings"] if (item["deviceId"], item["code"]) == identity), None)
if binding is not None:
slot = binding["slot"]
key = self.favorites().slots[slot].key if slot < 3 else document["slots"][slot - 3]
capture(key, timestamp)
return
if learning is not None:
if timestamp < learning["started"]:
return
openpilot/starpilot/controllers/runtime.py
+7 -2
View File
@@ -10,12 +10,13 @@ LOGGER = logging.getLogger(__name__)
class ControllerRuntime:
def __init__(self, params, *, actions, favorites, invoke_favorite, authority=None, reader=None, socket_path=None):
def __init__(self, params, *, actions, favorites, invoke_favorite, authority=None, reader=None, socket_path=None, input_capture=lambda: None):
self.authority = authority if authority is not None else LiveContextSource(params)
self.reader = reader if reader is not None else InputReader()
self.owner = ControllerOwner(params, parked=self.authority.parked, actions=actions, favorites=favorites,
invoke_favorite=invoke_favorite)
self.service = ControllerService(self.owner, socket_path)
self.input_capture = input_capture
self._closed = False
self._next_warm = 0.0
self._warm_until = time.monotonic() + 2.0
@@ -37,8 +38,12 @@ class ControllerRuntime:
self.owner.set_devices(self.reader.devices())
self.owner.tick()
# Process input before configuration so queued presses cannot activate a new binding.
capture = self.input_capture()
for press in presses:
self.owner.feed(press)
if capture is None:
self.owner.feed(press)
else:
self.owner.feed(press, capture=capture)
self.service.poll()
except Exception:
LOGGER.exception('Controller input stopped')
openpilot/starpilot/controllers/tests/test_capture.py
+55
View File
@@ -0,0 +1,55 @@
"""Native modal input consumes a complete batch without invoking vehicle actions."""
from unittest.mock import Mock
from openpilot.starpilot.controllers.input import Press
from openpilot.starpilot.controllers.tests.test_owner import TestControllerOwner as OwnerFixture
from openpilot.starpilot.controllers.tests.test_runtime import TestControllerRuntime as RuntimeFixture
def test_modal_resolves_binding_without_action_or_learning():
fixture = OwnerFixture()
fixture.setUp()
try:
fixture.configure()
fixture.learn(3)
fixture.now += 1_000_000
capture = Mock()
press = Press(fixture.device['id'], 42, fixture.now)
fixture.owner.feed(press, capture=capture)
capture.assert_called_once_with('safe.action', fixture.now)
assert fixture.calls == 0 and not fixture.favorite_calls
fixture.owner.feed(press, capture=capture)
assert capture.call_count == 1
fixture.owner.feed(Press('unknown', 42, fixture.now), capture=capture)
fixture.owner.feed(Press(fixture.device['id'], 42, fixture.now - 260_000_000), capture=capture)
assert capture.call_count == 1
fixture.now += 1_000_000
fixture.owner.feed(Press(fixture.device['id'], 42, fixture.now))
assert fixture.calls == 1
fixture.owner.action({'operation': 'learn', 'revision': fixture.owner.snapshot()['revision'], 'slot': 4})
fixture.now += 1_000_000
before = fixture.path.read_bytes()
fixture.owner.feed(Press(fixture.device['id'], 55, fixture.now), capture=capture)
assert fixture.path.read_bytes() == before
capture.assert_called_with(None, fixture.now)
finally:
fixture.doCleanups()
def test_batch_capture_is_stable_when_first_press_closes_modal():
fixture = RuntimeFixture()
runtime, _, reader, owner, _ = fixture.runtime()
try:
capture = Mock()
current: list[Mock | None] = [capture]
runtime.input_capture = lambda: current[0]
reader.poll.return_value = [Press('device', 1, 10), Press('device', 2, 20)]
def consume(_press, *, capture):
current[0] = None
owner.feed.side_effect = consume
runtime.poll()
assert owner.feed.call_count == 2
assert all(call.kwargs['capture'] is capture for call in owner.feed.call_args_list)
finally:
fixture.doCleanups()
openpilot/starpilot/parked_vehicle.py
+111
View File
@@ -0,0 +1,111 @@
"""Bounded receive-only vehicle observations while normal driving services are off.
This creates only existing CarState decoders and CAN parsers. It never creates a
CarInterface/CarController, fingerprints, initializes ECUs, or publishes CAN.
The caller must independently verify physical Panda noOutput and process state.
"""
from dataclasses import dataclass
import importlib
import math
import time
from openpilot.cereal import messaging
from opendbc.car.can_definitions import CanData
from opendbc.car.values import PLATFORMS
MAX_AGE_NS = 250_000_000
MAX_PACKETS = 32
MAX_FRAMES = 512
DWELL_NS = 1_000_000_000
@dataclass(frozen=True)
class ParkedSample:
source_boot_ns: int
received_mono_ns: int
buttons: tuple[tuple[str, bool], ...]
def stationary(state) -> bool:
values = (state.vEgo, state.vEgoRaw, state.wheelSpeeds.fl, state.wheelSpeeds.fr,
state.wheelSpeeds.rl, state.wheelSpeeds.rr)
return bool(state.canValid and not state.canTimeout and state.standstill and str(state.gearShifter) == 'park' and
not state.gasPressed and not state.cruiseState.enabled and
all(math.isfinite(v) and abs(v) <= .01 for v in values))
class ParkedVehicleReader:
"""Independent raw CAN subscriber with original parser checksum/counter checks."""
def __init__(self, cp):
platform = PLATFORMS.get(cp.carFingerprint)
if cp.notCar or platform is None or platform.__module__.split('.')[-2] != cp.brand:
raise ValueError('A recognized current vehicle identity is required')
# Detached copy prevents a UI/manager transition changing parser identity.
self.cp = cp.as_builder().as_reader()
state_type = importlib.import_module(f'opendbc.car.{cp.brand}.carstate').CarState
self.state = state_type(self.cp)
self.parsers = self.state.get_can_parsers(self.cp)
if not self.parsers or any(parser is None for parser in self.parsers.values()):
raise ValueError('Vehicle does not expose a complete read-only parser')
self.socket = messaging.sub_sock('can', conflate=False)
self.after_boot_ns = time.clock_gettime_ns(time.CLOCK_BOOTTIME)
self.last_source_ns = 0
self.sample: ParkedSample | None = None
self.since_ns: int | None = None
self.closed = False
def poll(self) -> tuple[ParkedSample, ...]:
if self.closed:
return ()
samples = []
for _ in range(MAX_PACKETS):
message = messaging.recv_one_or_none(self.socket)
if message is None:
return tuple(samples)
# Sample AFTER receiving: a publisher may commit a new packet during recv.
received = time.monotonic_ns()
boot = time.clock_gettime_ns(time.CLOCK_BOOTTIME)
stamp = int(message.logMonoTime)
if stamp <= self.after_boot_ns:
continue
if (not message.valid or stamp <= self.last_source_ns or not 0 <= boot - stamp <= MAX_AGE_NS or
len(message.can) > MAX_FRAMES):
raise ValueError('CAN observation is stale, invalid or out of order')
self.last_source_ns = stamp
packets = [(stamp, [CanData(int(frame.address), bytes(frame.dat), int(frame.src)) for frame in message.can])]
for parser in self.parsers.values():
parser.update(packets)
state = self.state.update(self.parsers)
state.canValid = all(parser.can_valid for parser in self.parsers.values())
state.canTimeout = any(parser.bus_timeout for parser in self.parsers.values())
self.state.out = state
# Parser can_valid intentionally tolerates a few failures and learns some
# frequencies. The game additionally requires fresh actual required frames;
# receiving unrelated CAN cannot keep a remembered Park/zero speed alive.
required = [message for parser in self.parsers.values() for message in parser.message_states.values() if not message.ignore_alive]
fresh = bool(required) and all(message.timestamps and
0 <= stamp - message.timestamps[-1] <= MAX_AGE_NS for message in required)
if not stationary(state) or not fresh:
self.sample = None
self.since_ns = None
continue
if self.sample is None or not 0 <= received - self.sample.received_mono_ns <= MAX_AGE_NS:
self.since_ns = received
self.sample = ParkedSample(stamp, received, tuple((str(e.type), bool(e.pressed)) for e in state.buttonEvents))
samples.append(self.sample)
raise ValueError('CAN observation exceeded the bounded drain')
def ready(self, now_mono_ns: int, now_boot_ns: int) -> bool:
sample = self.sample
return bool(not self.closed and sample is not None and self.since_ns is not None and
now_mono_ns - self.since_ns >= DWELL_NS and
0 <= now_mono_ns - sample.received_mono_ns <= MAX_AGE_NS and
0 <= now_boot_ns - sample.source_boot_ns <= MAX_AGE_NS)
def close(self) -> None:
self.closed = True
self.socket = None
self.sample = None
self.since_ns = None
self.parsers.clear()
openpilot/starpilot/tests/test_parked_vehicle.py
+138
View File
@@ -0,0 +1,138 @@
"""Receive-only native CAN parser proof, bounded input and physical freshness."""
from collections import deque
from types import SimpleNamespace as NS
from unittest.mock import Mock
import pytest
from openpilot.cereal import messaging
from openpilot.starpilot import parked_vehicle as vehicle
from openpilot.starpilot.ui.tests.test_ioniq6_alpha_request import stock_ioniq
from opendbc.can.packer import CANPacker
class CANFixture:
def __init__(self, monkeypatch):
self.now = 10_000_000_000
self.queue = deque()
monkeypatch.setattr(vehicle.time, 'monotonic_ns', lambda: self.now)
monkeypatch.setattr(vehicle.time, 'clock_gettime_ns', lambda _: self.now + 2_000_000_000)
monkeypatch.setattr(messaging, 'sub_sock', Mock())
monkeypatch.setattr(messaging, 'recv_one_or_none', lambda _: self.queue.popleft() if self.queue else None)
self.reader = vehicle.ParkedVehicleReader(stock_ioniq().as_reader())
self.packers = {bus: CANPacker(parser.dbc_name) for bus, parser in self.reader.parsers.items()}
# Normal CarState lazy subscriptions become explicit before generating data.
self.reader.state.update(self.reader.parsers)
self.park = next(value for value, text in self.reader.state.shifter_values.items() if text in ('P', 'PARK'))
self.drive = next(value for value, text in self.reader.state.shifter_values.items() if text in ('D', 'DRIVE'))
def event(self, *, gear=None, speed=0, button=0, omit=()):
frames = []
for bus, parser in self.reader.parsers.items():
for state in tuple(parser.message_states.values()):
if state.name in omit:
continue
values = {}
if state.name == self.reader.state.gear_msg_canfd:
values['GEAR'] = self.park if gear is None else gear
if state.name == 'WHEEL_SPEEDS':
values.update(dict.fromkeys(('WHL_SpdFLVal', 'WHL_SpdFRVal', 'WHL_SpdRLVal', 'WHL_SpdRRVal'), speed))
if state.name == 'CRUISE_BUTTONS':
values['CRUISE_BUTTONS'] = button
address, data, src = self.packers[bus].make_can_msg(state.name, parser.bus, values)
frames.append({'address': address, 'dat': data, 'src': src})
event = messaging.new_message('can', len(frames), valid=True)
event.logMonoTime = self.now + 2_000_000_000
event.can = frames
return event
def tick(self, **kwargs):
self.now += 10_000_000
self.queue.append(self.event(**kwargs).as_reader())
return self.reader.poll()
def warm(self):
for _ in range(150):
self.tick()
assert self.reader.ready(self.now, self.now + 2_000_000_000)
@pytest.fixture
def can(monkeypatch):
fixture = CANFixture(monkeypatch)
yield fixture
fixture.reader.close()
def test_actual_carstate_and_canparser_park_buttons_and_cleanup(can):
can.warm()
assert vehicle.stationary(can.reader.state.out)
found = []
for pressed in (0, 1, 1, 0, 2, 0):
found.extend(event for sample in can.tick(button=pressed) for event in sample.buttons)
assert ('accelCruise', True) in found and ('accelCruise', False) in found
assert ('decelCruise', True) in found and ('decelCruise', False) in found
can.reader.close()
assert can.reader.socket is None and not can.reader.parsers
assert not can.reader.ready(can.now, can.now + 2_000_000_000)
@pytest.mark.parametrize('fault', ['drive', 'speed', 'stale_gear', 'stale_wheels', 'source', 'receiver', 'future', 'invalid', 'order', 'backlog'])
def test_actual_decoder_refuses_changed_or_stale_physical_state(can, fault):
can.warm()
if fault in ('drive', 'speed'):
can.tick(**({'gear': can.drive} if fault == 'drive' else {'speed': 1.}))
elif fault in ('stale_gear', 'stale_wheels'):
name = can.reader.state.gear_msg_canfd if fault == 'stale_gear' else 'WHEEL_SPEEDS'
for _ in range(27):
can.tick(omit=(name,))
elif fault in ('source', 'receiver'):
now, boot = can.now, can.now + 2_000_000_000
assert not can.reader.ready(now + (260_000_000 if fault == 'receiver' else 0), boot + (260_000_000 if fault == 'source' else 0))
return
else:
can.now += 10_000_000
event = can.event()
if fault == 'future':
event.logMonoTime += 1
elif fault == 'invalid':
event.valid = False
elif fault == 'order':
event.logMonoTime = can.reader.last_source_ns
if fault == 'backlog':
for i in range(vehicle.MAX_PACKETS):
message = can.event()
message.logMonoTime = can.now + 2_000_000_000 - vehicle.MAX_PACKETS + i
can.queue.append(message.as_reader())
else:
can.queue.append(event.as_reader())
with pytest.raises(ValueError):
can.reader.poll()
return
assert not can.reader.ready(can.now, can.now + 2_000_000_000)
def test_receipt_clock_is_sampled_after_recv_not_before(can, monkeypatch):
can.warm()
def received(_):
if not can.queue:
return None
can.now += 1_000_000
event = can.queue.popleft().as_builder()
event.logMonoTime = can.now + 2_000_000_000
return event.as_reader()
monkeypatch.setattr(messaging, 'recv_one_or_none', received)
can.tick()
assert can.reader.ready(can.now, can.now + 2_000_000_000)
def test_constructor_never_creates_interface_controller_or_publisher(monkeypatch):
from opendbc.car.hyundai.interface import CarInterface
monkeypatch.setattr(CarInterface, '__init__', Mock(side_effect=AssertionError('No control interface in game')))
monkeypatch.setattr(messaging, 'PubMaster', Mock(side_effect=AssertionError('No publishing in game')))
monkeypatch.setattr(messaging, 'pub_sock', Mock(side_effect=AssertionError('No publishing in game')))
reader = vehicle.ParkedVehicleReader(stock_ioniq().as_reader())
reader.close()
with pytest.raises(ValueError):
vehicle.ParkedVehicleReader(NS(carFingerprint='unknown', notCar=False, brand='hyundai'))
openpilot/starpilot/ui/runtime_app.py
+10 -1
View File
@@ -1828,9 +1828,14 @@ class StarMainLayout(MainLayout):
toggle_items = self._large_panels[Destination.TOGGLES]._scroller._items
toggle_items.insert(2, text_item("Safe Mode", "N/A"))
toggle_items.insert(7, text_item("Right Hand Driving", "N/A"))
from openpilot.common.hardware import HARDWARE
from openpilot.starpilot.ui.starpilot_settings_adapter_large import StarPilotSettingsAdapterLarge
for destination, owner in self._large_panels.items():
self._large_panels[destination] = StarPilotSettingsAdapterLarge(owner, self.star.fonts, destination.value.title(), self._settings_back)
if destination == Destination.DEVELOPER and HARDWARE.get_device_type() in ('mici', 'tici', 'tizi'):
from openpilot.starpilot.ui.spacepilot.large import SpacePilotDeveloperAdapter
self._large_panels[destination] = SpacePilotDeveloperAdapter(owner, self.star.fonts, self._settings_back, self._exit_spacepilot)
else:
self._large_panels[destination] = StarPilotSettingsAdapterLarge(owner, self.star.fonts, destination.value.title(), self._settings_back)
self._network_bridge.presentation = StarPilotSettingsAdapterLarge(native_network, self.star.fonts, "Network", self._settings_back)
self.star.set_navigation(on_settings=lambda: self._set_current_layout(MainState.SETTINGS),
on_home=self._set_mode_for_state,
@@ -1840,6 +1845,10 @@ class StarMainLayout(MainLayout):
self.star._native_favorite_actions = self._native_favorites
self.star._publish_bookmark = self._on_bookmark_clicked
def _exit_spacepilot(self) -> None:
gui_app.pop_widgets_to(self, instant=True)
self._set_mode_for_state()
def _native_favorites(self):
from openpilot.selfdrive.ui.layouts.settings.settings import PanelType
return self.star.native_favorite_actions(
openpilot/starpilot/ui/spacepilot/README.md
+54 -35
View File
@@ -1,45 +1,64 @@
# SpacePilot
Native C4 Easter egg: in Developer settings, tap the **version number** five
times. Each tap must release within 450 ms without dragging; consecutive taps
must be 100–850 ms apart, all within three seconds. Navigation, scrolling,
touching another tile, loss of parked evidence, or a long press resets progress.
In native Developer settings, tap the **version number** five times. Both the
comma four and Big UI provide the same gesture and native game. Release each tap
within 450 ms without dragging; consecutive taps must be 100–850 ms apart, all
within three seconds. Scrolling, another row, navigation, a long press, or loss
of physical parked evidence resets progress.
Tap the playfield to launch/boost. The wheel's + and − events are read through an
independent, non-conflated `carState` subscriber while the game is admitted.
Release a wheel button once to arm it, then each press boosts once. Held buttons,
duplicate messages and repeated press events cannot continuously boost. Exit
returns to Developer settings; an unsafe exit immediately reveals the native
road/alert screen. Best score lasts for this game session.
Keep the vehicle in Park and disengaged while the preparation screen verifies
its state and stops driving services. Play begins only after those services are
confirmed stopped and every connected Panda reports healthy `noOutput`.
Tap to start, boost, and retry. With ignition on, the wheel's + and − are observed
through an independent receive-only CAN decoder. Release a wheel button once to
arm it; each following press boosts once. Held/repeated inputs do not repeat.
## Admission and lifecycle
Offroad with ignition off, use touch or a paired USB/Bluetooth controller. Enable
and configure the controller's existing **Increase Set Speed** / **Decrease Set
Speed** bindings in Controller settings. In SpacePilot those bindings only boost
the ship; no cruise request is dispatched. All controller input in the game's
input batch is consumed, even if the first press closes the game. Exiting restores
ordinary controller routing, subject to its original control authority checks.
Only the native C4 shell installs the version gesture. There is no environment
override for admission. Replay, projection, non-car devices and unknown state
are refused. One continuous second of fresh physical evidence is required:
Park, standstill, effectively zero ego/raw/all wheel speeds, valid CAN, ignition,
recognized connected Pandas, no Panda faults/control permission, no ACC, no
selfdrive or lateral/longitudinal control activity, and no active native alert.
Both source and receiver timestamps are checked with their original clocks.
The game closes on every failing check, producer gap, clock discontinuity or
wheel backlog. Safety is checked before drawing and before touch actions.
## Lifecycle and physical evidence
The game never writes vehicle messages, drive-mode preferences, manager state,
control permissions, process configuration or audio state. It does not start any
driving process. It uses the already-running, physically parked pipeline.
Onroad admission retains the original one-second fresh physical Park/standstill,
zero ego/raw/all wheel speed, valid CAN, disengaged ACC/controls, healthy connected
Panda, and no native alert requirements. Replay/projection, unknown state and
host overrides cannot admit the game. The version row owns only the gesture;
the page owns its own admission and cleanup state.
Force Offroad stops `card` and `selfdrived`, losing trustworthy gear/speed and
wheel input. The existing offroad physical-evidence fallback can confirm
ignition/noOutput but cannot provide these inputs. Therefore Force Offroad
cannot admit or sustain SpacePilot. No speculative parked process mode is added.
The game requests Offroad through the existing session-bound `DriveStateOwner`.
It does not falsify vehicle state, change manager process predicates or native
safety, initialize a vehicle interface/controller, send CAN, or alter alerts.
Normal `pandad` retains raw CAN reception offroad and selects `noOutput`. A
separate bounded subscriber instantiates only the current vehicle's normal
`CarState` and CAN parsers, retaining checksum/counter/timeout checks. Required
CAN frames additionally must remain within 250 ms; unrelated packets cannot keep
a remembered Park or zero speed alive. Unsupported or incomplete decoding fails
closed. Current live-session `carParams` is required for ignition-on play.
The game waits for actual `card`, `selfdrived`, `controlsd`, `modeld`, and
`dmonitoringmodeld` shutdown. Fresh manager/device/Panda source and receipt clocks,
physical noOutput, independent Park evidence, clock continuity and input bounds
remain mandatory. With ignition off, fresh healthy noOutput and actual stopped
processes provide the existing physical offroad premise; wheel input is disabled.
A mode request alone never proves that the car is parked. Changing ignition,
mode, gear, speed, engagement, process state, faults, source freshness or clock
continuity ends the game. Entry/shutdown times out after eight seconds.
A successful onroad entry **leaves Offroad selected when the game exits or the UI
closes**. Returning to driving requires an explicit existing drive-mode selection
(Onroad or Auto), which performs the normal offroad/onroad startup. The game never
restores driving services automatically. This is the existing manager-session
request, not a new persistent safety mode across manager/system restarts.
## Validation boundary
`test_spacepilot_state.py` covers refusal, original source/receipt clocks, dwell,
gesture timing, input edges, simulation, scoring and bounded storage.
`test_spacepilot_native.py` covers actual cereal schemas, native widget events,
return-to-road routing, lifecycle cleanup and both raylib gradient signatures.
Native GL screenshots and real isolated button IPC are host fixtures, not
physical vehicle proof. Physical C4 full-pipeline CPU/GPU/frame-time and actual
wheel validation still require an authorized, exclusively leased desk device.
Never infer road readiness or deployment from these tests.
Tests cover physical admission, stale source/receipt clocks, process/noOutput
transitions, the actual drive-state owner, touch/wheel/controller isolation,
normal CAN decoding and refusal, both native version gestures, bounded simulation
and cleanup. Native GL previews and host CPU/RSS/frame-time measurements use
isolated synthetic telemetry. They do not establish physical C4/C3 performance,
actual wheel/controller hardware behavior, full-system headroom or deployment.
No road-device installation, restart, or driving test is part of host validation.
openpilot/starpilot/ui/spacepilot/guard.py
+55 -3
View File
@@ -18,13 +18,14 @@ class Witness:
boot_ns: int
stamps: tuple[int, ...]
buttons: tuple[tuple[str, bool], ...]
mode: str = 'onroad'
def read_witness(ui: Any, mono_ns: int, boot_ns: int, *, c4: bool) -> Witness | None:
"""Original message and receipt clocks must both be fresh; Panda uses BOOTTIME.
The supported onroad producers must remain running. Force Offroad stops them,
so it is neither an admission signal nor a resource-management option here.
This proves the pre-transition state only. Offroad gameplay uses independent
CAN plus actual process shutdown and physical noOutput evidence below.
"""
try:
if (not c4 or mono_ns <= 0 or boot_ns <= 0 or not ui.started or ui.CP is None or ui.CP.notCar or
@@ -73,7 +74,7 @@ class ParkGate:
if witness is None:
self.reset()
return False
if previous is not None and (not 0 <= witness.mono_ns - previous.mono_ns <= FAST_AGE_NS or
if previous is not None and (witness.mode != previous.mode or not 0 <= witness.mono_ns - previous.mono_ns <= FAST_AGE_NS or
abs((witness.boot_ns - witness.mono_ns) - (previous.boot_ns - previous.mono_ns)) > RESUME_SKEW_NS or
any(new < old for new, old in zip(witness.stamps, previous.stamps, strict=True))):
self.reset()
@@ -82,3 +83,54 @@ class ParkGate:
self.since = witness.mono_ns
self.ready = witness.mono_ns - self.since >= DWELL_NS
return self.ready
DRIVING_PROCESSES = frozenset(('card', 'selfdrived', 'controlsd', 'modeld', 'dmonitoringmodeld'))
def read_offroad_witness(ui: Any, mono_ns: int, boot_ns: int, *, native_device: bool, vehicle=None, preparing: bool = False) -> Witness | None:
"""Require actual quiet processes and healthy physical noOutput, not a mode request.
preparing=True may open only the verification screen, never gameplay. Its
independent decoder must prove Park before the full witness can be admitted.
Ignition-off uses the same physical parked premise as the existing native
parked-action owner. With ignition on, a live independent CAN decoder must
continue proving Park/zero speed; a remembered pre-transition value is invalid.
"""
try:
if (not native_device or ui.started or ui.projection_read_only or ui.replay_clock is not None or
getattr(ui.sm, 'replay_sample', None) is not None or not ui.params.get_bool('IsOffroad')):
return None
sm = ui.sm
stamps = []
for name in ('deviceState', 'pandaStates', 'managerState'):
ttl = PANDA_TTL_NS if name == 'pandaStates' else SLOW_AGE_NS
stamp, received = int(sm.logMonoTime[name]), float(sm.recv_time[name])
now = boot_ns if name == 'pandaStates' else mono_ns
if (not sm.valid[name] or not sm.alive[name] or stamp <= 0 or not 0 <= now - stamp <= ttl or
not math.isfinite(received) or not 0 <= mono_ns - int(received * 1e9) <= ttl):
return None
stamps.append(stamp)
if sm['deviceState'].started or str(sm['deviceState'].thermalStatus) != 'ok':
return None
processes = tuple(sm['managerState'].processes)
names = [p.name for p in processes]
if len(names) != len(set(names)) or not DRIVING_PROCESSES.issubset(names):
return None
if any(p.running for p in processes if p.name in DRIVING_PROCESSES):
return None
pandas = tuple(sm['pandaStates'])
if not pandas:
return None
for panda in pandas:
if (str(panda.pandaType) == 'unknown' or str(panda.safetyModel) != 'noOutput' or panda.controlsAllowed or
panda.heartbeatLost or panda.safetyRxChecksInvalid or str(panda.faultStatus) != 'none' or len(panda.faults)):
return None
ignition = any(p.ignitionLine or p.ignitionCan for p in pandas)
if ignition and not preparing:
if (any(str(p.harnessStatus) == 'notConnected' for p in pandas) or vehicle is None or
not vehicle.ready(mono_ns, boot_ns)):
return None
return Witness(mono_ns, boot_ns, tuple(stamps), (), 'offroad-check' if ignition and preparing else 'offroad')
except (AttributeError, KeyError, TypeError, ValueError, OverflowError, OSError, RuntimeError):
return None
openpilot/starpilot/ui/spacepilot/large.py
+113
View File
@@ -0,0 +1,113 @@
"""The native Big UI Developer version row, sharing the parked game and gesture."""
import math
import time
import pyray as rl
from openpilot.selfdrive.ui.ui_state import ui_state
from openpilot.starpilot.ui.feature_settings import DETAIL_SIZE, LABEL_SIZE
from openpilot.starpilot.ui.feature_settings_state import FEATURE_ROW_HEIGHT, feature_row_top
from openpilot.starpilot.ui.presentation import FontRole
from openpilot.starpilot.ui.starpilot_settings_adapter_large import StarPilotSettingsAdapterLarge
from openpilot.starpilot.ui.spacepilot.guard import ParkGate
from openpilot.starpilot.ui.spacepilot.native import SpacePilotPage, live_witness
from openpilot.starpilot.ui.spacepilot.state import FiveTaps
from openpilot.system.ui.lib.application import gui_app
from openpilot.system.ui.widgets.list_view import text_item
class SpacePilotDeveloperAdapter(StarPilotSettingsAdapterLarge):
def __init__(self, panel, fonts, on_back, on_unsafe):
super().__init__(panel, fonts, 'Developer', on_back)
self.version = text_item('Version', lambda: ui_state.params.get('Version') or 'N/A')
panel._scroller._items.insert(0, self.version)
self.on_unsafe = on_unsafe
self.gate = ParkGate()
self.taps = FiveTaps()
self.press = None
self.version_touch = False
def _reset_gesture(self):
self.taps.reset()
self.press = None
self.version_touch = False
def show_event(self):
self._reset_gesture()
self.gate.reset()
super().show_event()
def hide_event(self):
self._reset_gesture()
self.gate.reset()
super().hide_event()
def _update_state(self):
super()._update_state()
if not self.active or not self.enabled or not self.gate.update(live_witness()):
self._reset_gesture()
def _version_bounds(self):
state = self.snapshot()
if state.scroll or state.editor or not state.rows or state.rows[0].key != str(id(self.version)):
return rl.Rectangle(0, 0, 0, 0)
left = 520 if state.sidebar_expanded else 20
row = state.rows[0]
detail = self.view._elide(row.value, FontRole.NORMAL, DETAIL_SIZE, 2094 - left - 55)
lt, lb = self.view.fonts.vertical_ink(row.label, FontRole.NORMAL, LABEL_SIZE)
dt, db = self.view.fonts.vertical_ink(detail, FontRole.NORMAL, DETAIL_SIZE)
height = lb - lt + 12 + db - dt
top = feature_row_top(state) + (FEATURE_ROW_HEIGHT - height) / 2 + lb - lt + 12
return rl.Rectangle(self.origin[0] + left + 55, self.origin[1] + top - 8,
self.view.fonts.measure(detail, FontRole.NORMAL, DETAIL_SIZE).width, db - dt + 16)
def _process_mouse_events(self):
if not self._touch_valid():
self._reset_gesture()
for event in gui_app.mouse_events:
if event.slot == 0 and event.left_pressed and not rl.check_collision_point_rec(event.pos, self._version_bounds()):
self._reset_gesture()
super()._process_mouse_events()
def _handle_mouse_press(self, pos):
self.version_touch = bool(rl.check_collision_point_rec(pos, self._version_bounds()))
self.press = None
if self.version_touch:
self.input.cancel()
if self.view.reset(self.snapshot()):
self.taps.reset()
self.version_touch = False
return
self.taps.reset()
super()._handle_mouse_press(pos)
def _handle_mouse_event(self, mouse_event):
event = mouse_event
if not self.version_touch:
super()._handle_mouse_event(event)
return
if event.left_pressed:
self.press = event
elif self.press is not None:
if (not 0 <= event.t - self.press.t <= .45 or
math.hypot(event.pos.x - self.press.pos.x, event.pos.y - self.press.pos.y) > 24 or
not self._touch_valid()):
self.press = None
self.taps.reset()
elif event.left_released:
self.press = None
if (not rl.check_collision_point_rec(event.pos, self._version_bounds()) or
not self.gate.update(live_witness())):
self.taps.reset()
elif self.taps.tap(time.monotonic()):
gui_app.mouse_events.clear()
gui_app.push_widget(SpacePilotPage(self.gate, self.on_unsafe))
def _handle_mouse_release(self, pos):
if not self.version_touch:
super()._handle_mouse_release(pos)
def _emit(self, request):
self._reset_gesture()
super()._emit(request)
openpilot/starpilot/ui/spacepilot/native.py
+62 -72
View File
@@ -1,4 +1,4 @@
"""C4 native page and Developer version gesture. All vehicle access is read-only."""
"""Native game and version gesture using the supported parked lifecycle owner."""
from collections.abc import Callable
import math
@@ -6,14 +6,17 @@ import time
import pyray as rl
import openpilot.cereal.messaging as messaging
from openpilot.common.hardware import HARDWARE
from openpilot.common.swaglog import cloudlog
from openpilot.selfdrive.ui.mici.widgets.button import BigButton
from openpilot.selfdrive.ui.ui_state import ui_state
from openpilot.starpilot.ui.spacepilot.guard import FAST_AGE_NS, ParkGate, Witness, read_witness
from openpilot.starpilot.ui.spacepilot.render import EXIT, SpaceRenderer
from openpilot.starpilot.ui.spacepilot.state import FiveTaps, Flight, WheelInput
from openpilot.starpilot.ui.spacepilot.guard import ParkGate, Witness, read_offroad_witness, read_witness
from openpilot.starpilot.ui.spacepilot.session import GameSession, Stage
from openpilot.starpilot.drive_state.owner import DriveStateOwner
from openpilot.starpilot.storage import starpilot_storage_root
from openpilot.starpilot.ui.spacepilot.render import EXIT, SpaceRenderer, canvas_rect
from openpilot.starpilot.favorites.actions import INCREASE_SPEED, DECREASE_SPEED
from openpilot.starpilot.ui.spacepilot.state import FiveTaps, Flight
from openpilot.system.ui.lib.application import MouseEvent, MousePos, gui_app
from openpilot.system.ui.widgets import Widget
@@ -25,69 +28,35 @@ def live_witness() -> Witness | None:
after = time.monotonic_ns()
if not 0 <= after - before <= 1_000_000:
return None
return read_witness(ui_state, after, boot, c4=HARDWARE.get_device_type() == 'mici')
native_device = HARDWARE.get_device_type() in ('mici', 'tici', 'tizi')
if ui_state.started:
return read_witness(ui_state, after, boot, c4=native_device)
return read_offroad_witness(ui_state, after, boot, native_device=native_device, preparing=True)
except (AttributeError, OSError, ValueError):
return None
class WheelReader:
"""Separate non-conflated subscription keeps 100 Hz press/release edges at 60 Hz UI.
It exists only while admitted. It never consumes another subscriber's input or
publishes any messages. Dropping the sole socket reference closes native IPC.
"""
def __init__(self, after_ns: int):
self.socket = messaging.sub_sock('carState', conflate=False)
self.input = WheelInput(after_ns)
def poll(self, now_ns: int) -> bool:
fire = False
for _ in range(32):
message = messaging.recv_one_or_none(self.socket)
if message is None:
return fire
stamp = int(message.logMonoTime)
if stamp <= self.input.stamp:
continue
cs = message.carState
if (not message.valid or not 0 <= now_ns - stamp <= FAST_AGE_NS or not cs.canValid or cs.canTimeout or
str(cs.gearShifter) != 'park' or not cs.standstill or
any(not math.isfinite(v) or abs(v) > .01 for v in (cs.vEgo, cs.vEgoRaw))):
raise ValueError('Wheel observation is no longer trustworthy')
events = tuple((str(event.type), bool(event.pressed)) for event in cs.buttonEvents)
fire = self.input.update(stamp, events, stamp / 1e9) or fire
raise ValueError('Wheel input backlog exceeded the bounded drain')
def close(self) -> None:
self.socket = None
self.input = WheelInput()
class SpacePilotPage(Widget):
def __init__(self, gate: ParkGate, on_unsafe: Callable[[], None]):
super().__init__()
self.gate = gate
self.gate = ParkGate()
self.gate.since, self.gate.previous, self.gate.ready = gate.since, gate.previous, gate.ready
self.session = GameSession(ui_state, DriveStateOwner(ui_state.params, starpilot_storage_root() / 'drive-state'),
native_device=HARDWARE.get_device_type() in ('mici', 'tici', 'tizi'))
self.on_unsafe = on_unsafe
self.flight = Flight(time.monotonic_ns())
self.renderer = SpaceRenderer()
self.wheel: WheelReader | None = None
self.controller_after_ns = time.monotonic_ns()
self.controller_last_ns = 0
self.last_tick = 0.
self.closed = True
self.touch_held = self.exit_pressed = False
def show_event(self) -> None:
super().show_event()
witness = live_witness()
if not self.gate.update(witness) or witness is None:
# push_widget invokes show_event before enabling this page; defer stack
# mutation to its first tick, and never acquire a socket after refusal.
self.closed = False
else:
self.closed = False
try:
self.wheel = WheelReader(witness.mono_ns)
except (OSError, RuntimeError):
self.gate.reset()
self.closed = False
if not self.gate.update(live_witness()):
self.session.close('Entry evidence changed')
self.last_tick = time.monotonic()
self.touch_held = self.exit_pressed = False
gui_app.add_nav_stack_tick(self._tick)
@@ -95,32 +64,34 @@ class SpacePilotPage(Widget):
def hide_event(self) -> None:
self.closed = True
gui_app.remove_nav_stack_tick(self._tick)
if self.wheel is not None:
self.wheel.close()
self.wheel = None
self.session.close()
self.flight.close()
self.gate.reset()
self.touch_held = self.exit_pressed = False
super().hide_event()
def _safe(self) -> bool:
return bool(not self.closed and gui_app.get_active_widget() is self and self.gate.update(live_witness()))
return bool(not self.closed and gui_app.get_active_widget() is self and self.session.playable())
def _tick(self) -> None:
if self.closed:
return
if not self._safe() or self.wheel is None:
if gui_app.get_active_widget() is not self:
self._exit(unsafe=True)
return
fire = self.session.step()
if self.session.stage == Stage.CLOSED:
cloudlog.info('SpacePilot ended: %s', self.session.reason)
self._exit(unsafe=True)
return
now = time.monotonic()
try:
if self.wheel.poll(time.monotonic_ns()):
self.flight.act(now)
except (OSError, RuntimeError, ValueError, AttributeError):
self._exit(unsafe=True)
return
self.flight.update(now - self.last_tick)
if fire:
self.flight.act(now)
if self.session.stage == Stage.PLAYING:
self.flight.update(now - self.last_tick)
self.last_tick = now
if self.session.stage != Stage.PLAYING:
self.controller_after_ns = time.monotonic_ns()
def _exit(self, *, unsafe: bool) -> None:
if self.closed:
@@ -132,19 +103,35 @@ class SpacePilotPage(Widget):
elif gui_app.get_active_widget() is self:
gui_app.pop_widget()
def capture_controller(self, key: str | None, timestamp_ns: int) -> None:
# The shared input owner validates identity, press/release, duplicate and age.
# Consume every action while this page owns the batch, including during exit.
if (key not in (INCREASE_SPEED, DECREASE_SPEED) or timestamp_ns <= self.controller_after_ns or
timestamp_ns - self.controller_last_ns < 120_000_000):
return
if not self._safe():
if self.session.stage == Stage.PLAYING:
self._exit(unsafe=True)
return
self.controller_last_ns = timestamp_ns
self.flight.act(time.monotonic())
def _in_exit(self, pos: MousePos) -> bool:
x, y, w, h = EXIT
return bool(rl.check_collision_point_rec(pos, rl.Rectangle(self.rect.x + x * self.rect.width / 1072,
self.rect.y + y * self.rect.height / 480, w * self.rect.width / 1072, h * self.rect.height / 480)))
rect = canvas_rect(self.rect)
return bool(rl.check_collision_point_rec(pos, rl.Rectangle(rect.x + x * rect.width / 1072,
rect.y + y * rect.height / 480, w * rect.width / 1072, h * rect.height / 480)))
def _handle_mouse_press(self, pos: MousePos) -> None:
if self.touch_held:
return
self.touch_held = True
self.exit_pressed = self._in_exit(pos)
if self.session.stage != Stage.PLAYING:
return
if not self._safe():
self._exit(unsafe=True)
return
self.exit_pressed = self._in_exit(pos)
if not self.exit_pressed:
self.flight.act(time.monotonic())
@@ -157,15 +144,18 @@ class SpacePilotPage(Widget):
self.exit_pressed = False
def _render(self, rect: rl.Rectangle) -> None:
# The check also protects rendering when another native caller renders us
# directly instead of through the application's navigation tick.
if not self._safe():
if self.closed or gui_app.get_active_widget() is not self:
self._exit(unsafe=True)
return
try:
self.renderer.render(self.flight, rect, wheel_ready=bool(self.wheel and self.wheel.input.armed))
if self.session.stage == Stage.PLAYING:
if not self._safe():
self._exit(unsafe=True)
return
self.renderer.render(self.flight, rect, wheel_ready=bool(self.session.wheel.armed), wheel_available=self.session.vehicle is not None)
else:
self.renderer.preparing(rect, stopping=self.session.stage == Stage.STOPPING)
except Exception:
# An optional Easter egg must not take the driving/alert UI down with it.
cloudlog.exception('SpacePilot render failed')
self._exit(unsafe=True)
openpilot/starpilot/ui/spacepilot/render.py
+34 -6
View File
@@ -18,6 +18,12 @@ GOLD = rl.Color(255, 207, 121, 255)
EXIT = (938., 10., 118., 48.)
def canvas_rect(rect: rl.Rectangle) -> rl.Rectangle:
scale = min(rect.width / 1072, rect.height / 480)
return rl.Rectangle(rect.x + (rect.width - 1072 * scale) / 2, rect.y + (rect.height - 480 * scale) / 2,
1072 * scale, 480 * scale)
class SpaceRenderer:
def __init__(self):
rng = random.Random(5683)
@@ -75,7 +81,27 @@ class SpaceRenderer:
rl.draw_line_ex((-4, -7), (17, -1), 2, CYAN)
rl.rl_pop_matrix()
def render(self, game: Flight, rect: rl.Rectangle, *, wheel_ready: bool) -> None:
def preparing(self, rect: rl.Rectangle, *, stopping: bool) -> None:
rl.draw_rectangle_rec(rect, INK)
rect = canvas_rect(rect)
rl.rl_push_matrix()
rl.rl_translatef(rect.x, rect.y, 0)
rl.rl_scalef(rect.width / 1072, rect.height / 480, 1)
try:
rl.draw_rectangle_gradient_v(0, 0, 1072, 480, rl.Color(16, 20, 48, 255), INK)
self.text('SpacePilot', 536, 109, 62, centered=True, bold=True)
self.text('Entering parked game mode' if stopping else 'Checking physical Park', 536, 211, 27, CYAN, centered=True)
self.text('Waiting for driving services to stop' if stopping else 'Keep the car in Park and release the wheel buttons',
536, 265, 21, MUTED, centered=True)
self.text('Driving services stay off until you select Onroad or Auto again.', 536, 362, 19, MUTED, centered=True)
self.panel(*EXIT, rl.Color(30, 43, 62, 255))
self.text('EXIT', 997, 19, 22, centered=True, bold=True)
finally:
rl.rl_pop_matrix()
def render(self, game: Flight, rect: rl.Rectangle, *, wheel_ready: bool, wheel_available: bool = True) -> None:
rl.draw_rectangle_rec(rect, INK)
rect = canvas_rect(rect)
rl.rl_push_matrix()
rl.rl_translatef(rect.x, rect.y, 0)
rl.rl_scalef(rect.width / 1072, rect.height / 480, 1)
@@ -110,16 +136,18 @@ class SpaceRenderer:
self.text('EXIT', 997, 19, 22, centered=True, bold=True)
rl.draw_rectangle(0, 439, 1072, 41, INK)
rl.draw_line(0, 438, 1072, 438, rl.Color(46, 66, 88, 255))
self.text('P / STATIONARY', 22, 450, 16, CYAN)
self.text('TAP OR + / – TO BOOST', 1050 - measure_text_cached(gui_app.font(), 'TAP OR + / – TO BOOST', 16).x, 450, 16, MUTED)
self.text('P / DRIVING SERVICES OFF', 22, 450, 16, CYAN)
hint = 'TAP OR + / – TO BOOST' if wheel_available else 'TOUCH OR PAIRED CONTROLLER'
self.text(hint, 1050 - measure_text_cached(gui_app.font(), hint, 16).x, 450, 16, MUTED)
if game.phase == Phase.READY:
self.ship(247, 234 + math.sin(game.age * 2.2) * 8, -8 + math.sin(game.age * 1.4) * 4, game.age, scale=2.7)
self.text('A LITTLE ESCAPE', 675, 100, 18, CYAN, centered=True)
self.text('SpacePilot', 675, 139, 62, centered=True, bold=True)
self.text('Thread the gates. Chase the stars.', 675, 224, 22, MUTED, centered=True)
self.panel(395, 286, 560, 60, rl.Color(38, 80, 103, 255))
self.text('Press +/– or tap to start', 675, 300, 27, centered=True, bold=True)
self.text('Release each wheel button once to arm it' if not wheel_ready else 'One press. One boost. Find your rhythm.',
self.text('Press +/– or tap to start' if wheel_available else 'Tap to start', 675, 300, 27, centered=True, bold=True)
self.text(('Release each wheel button once to arm it' if not wheel_ready else 'One press. One boost. Find your rhythm.')
if wheel_available else 'Controller: use your mapped +/– buttons.',
675, 369, 18, MUTED, centered=True)
elif game.phase == Phase.OVER:
self.panel(326, 99, 450, 306, rl.Color(10, 19, 37, 245))
@@ -127,6 +155,6 @@ class SpaceRenderer:
self.text(f'{game.score:02d}', 551, 147, 73, centered=True, bold=True)
self.text(f'GATES CLEARED / BEST {game.best:02d}', 551, 244, 19, MUTED, centered=True)
self.panel(356, 301, 390, 67, rl.Color(38, 80, 103, 255))
self.text('Press +/– or tap to retry', 551, 321, 23, centered=True, bold=True)
self.text('Press +/– or tap to retry' if wheel_available else 'Tap to retry', 551, 321, 23, centered=True, bold=True)
finally:
rl.rl_pop_matrix()
openpilot/starpilot/ui/spacepilot/session.py
+166
View File
@@ -0,0 +1,166 @@
"""Parked game lifecycle: observe, request supported Offroad, then verify shutdown.
The game never starts or stops a managed process itself. A successful onroad
entry leaves the existing DriveStateOwner in Offroad even after game exit; only
a later explicit drive-state selection can start driving services again.
"""
from enum import StrEnum
import time
from openpilot.starpilot.parked_vehicle import ParkedVehicleReader
from openpilot.starpilot.ui.spacepilot.guard import ParkGate, read_offroad_witness, read_witness
from openpilot.starpilot.ui.spacepilot.state import WheelInput
TRANSITION_NS = 8_000_000_000
class Stage(StrEnum):
VERIFYING = 'verifying'
STOPPING = 'stopping'
PLAYING = 'playing'
CLOSED = 'closed'
class GameSession:
def __init__(self, ui, drive_owner, *, native_device: bool, reader_factory=ParkedVehicleReader):
self.ui, self.drive_owner, self.native_device = ui, drive_owner, native_device
self.stage = Stage.VERIFYING
self.started_ns = time.monotonic_ns()
self.requested_offroad = False
self.vehicle = None
self.wheel = WheelInput()
self.gate = ParkGate()
self.reason = ''
self._reader_factory = reader_factory
self._was_ignition = None
self._request_revision = None
self._clock_offset = None
def _clocks(self):
before = time.monotonic_ns()
boot = time.clock_gettime_ns(time.CLOCK_BOOTTIME)
mono = time.monotonic_ns()
if not 0 <= mono - before <= 1_000_000:
raise ValueError('Waiting for a stable device clock')
offset = boot - mono
if self._clock_offset is not None and abs(offset - self._clock_offset) > 5_000_000:
raise ValueError('Device resumed; re-enter SpacePilot')
self._clock_offset = offset
return mono, boot
def _ignition(self):
return any(p.ignitionLine or p.ignitionCan for p in self.ui.sm['pandaStates'])
def _onroad_witness(self):
mono, boot = self._clocks()
return read_witness(self.ui, mono, boot, c4=self.native_device)
def step(self) -> bool:
"""Return one debounced boost only after actual process/noOutput admission."""
if self.stage == Stage.CLOSED:
return False
try:
mono, boot = self._clocks()
ignition = self._ignition()
if self._was_ignition is not None and ignition != self._was_ignition:
# Never reuse a parser or old admission across a physical ignition cycle.
raise ValueError('Vehicle ignition changed')
self._was_ignition = ignition
if self.stage != Stage.PLAYING and mono - self.started_ns > TRANSITION_NS:
raise ValueError('Driving services did not finish entering parked game mode')
if self.vehicle is None and ignition:
sm = self.ui.sm
if (not sm.valid['carParams'] or sm.recv_frame['carParams'] <= self.ui.started_frame or
not 0 < int(sm.logMonoTime['carParams']) <= mono):
raise ValueError('Current vehicle identity is unavailable')
self.vehicle = self._reader_factory(sm['carParams'])
samples = self.vehicle.poll() if self.vehicle is not None else ()
mono, boot = self._clocks()
vehicle_ready = bool(self.vehicle is not None and self.vehicle.ready(mono, boot))
if self.stage in (Stage.STOPPING, Stage.PLAYING) and ignition and not vehicle_ready:
raise ValueError('Physical Park or fresh CAN was lost')
if self.stage == Stage.VERIFYING and self.ui.started:
witness = self._onroad_witness()
if witness is None:
raise ValueError('Park and disengage before entering SpacePilot')
if not self.gate.update(witness) or not vehicle_ready:
return False
state = self.drive_owner.snapshot()
if not state.available:
raise ValueError('Drive-state owner is unavailable')
def parked_now():
now, boot_now = self._clocks()
return bool(self.stage == Stage.VERIFYING and self._onroad_witness() is not None and
self.vehicle is not None and self.vehicle.ready(now, boot_now))
result = self.drive_owner.request('offroad', expected_revision=state.revision,
authorized=parked_now, override_allowed=parked_now)
if str(result.mode) != 'offroad' or not result.available:
raise ValueError('Offroad request was not accepted')
self._request_revision = result.revision
self.requested_offroad = True
self.stage = Stage.STOPPING
self.started_ns = mono
self.gate.reset()
self.wheel = WheelInput()
return False
if self.requested_offroad:
state = self.drive_owner.snapshot()
if not state.available or str(state.mode) != 'offroad' or state.revision != self._request_revision:
raise ValueError('Drive mode changed; SpacePilot has ended')
if self.stage == Stage.STOPPING and self.ui.started:
# No game/input until all normal services have actually stopped. Keep
# the original fresh physical admission while they are still onroad.
if self._onroad_witness() is None or not vehicle_ready:
raise ValueError('Physical parked evidence changed during shutdown')
return False
witness = read_offroad_witness(self.ui, mono, boot, native_device=self.native_device, vehicle=self.vehicle)
ready = self.gate.update(witness)
if self.stage == Stage.PLAYING and not ready:
raise ValueError('Parked game evidence is no longer trustworthy')
if not ready:
return False
if self.stage != Stage.PLAYING:
self.stage = Stage.PLAYING
# Pre-admission presses/releases cannot arm a wheel action.
self.wheel = WheelInput(self.vehicle.sample.source_boot_ns if self.vehicle is not None else 0)
return False
fire = False
for sample in samples:
fire = self.wheel.update(sample.source_boot_ns, sample.buttons, mono / 1e9) or fire
return fire
except (AttributeError, KeyError, OSError, RuntimeError, TypeError, ValueError, ImportError) as exc:
self.close(str(exc))
return False
def playable(self) -> bool:
"""Touch/controller/render must revalidate, never rely on last frame alone."""
if self.stage != Stage.PLAYING:
return False
try:
mono, boot = self._clocks()
if self._ignition() != self._was_ignition:
return False
if self.requested_offroad:
state = self.drive_owner.snapshot()
if not state.available or str(state.mode) != 'offroad' or state.revision != self._request_revision:
return False
return self.gate.update(read_offroad_witness(self.ui, mono, boot, native_device=self.native_device, vehicle=self.vehicle))
except (AttributeError, KeyError, OSError, RuntimeError, TypeError, ValueError):
return False
def close(self, reason: str = '') -> None:
self.reason = reason
self.stage = Stage.CLOSED
if self.vehicle is not None:
self.vehicle.close()
self.vehicle = None
self.gate.reset()
self.wheel = WheelInput()
# Deliberately no request(Auto/Onroad): closing/crashing must not restart
# driving services or restore engagement processing behind a game button.
openpilot/starpilot/ui/tests/test_spacepilot_large.py
+83
View File
@@ -0,0 +1,83 @@
"""Real native gesture dispatch in Big UI, including canceled and moving rows."""
from types import SimpleNamespace as NS
from unittest.mock import Mock
import pyray as rl
import pytest
from openpilot.starpilot.ui.spacepilot import large, native
from openpilot.starpilot.ui.presentation import Profile
from openpilot.starpilot.ui.tests.test_spacepilot_state import mature_gate, parked_ui, read
from openpilot.system.ui.lib.application import MouseEvent, MousePos, gui_app
@pytest.fixture
def adapter(monkeypatch):
monkeypatch.setattr(gui_app, 'font', lambda *_: rl.Font())
fonts = NS(profile=Profile.LARGE, vertical_ink=lambda _t, _r, size: (0, size),
measure=lambda text, _r, size: NS(width=len(text) * size / 2))
panel = NS(_scroller=NS(_items=[]), show_event=Mock(), hide_event=Mock(), _update_state=Mock())
ui = NS(params=NS(get=lambda _: '0.10.3'))
monkeypatch.setattr(large, 'ui_state', ui)
monkeypatch.setattr(native, 'ui_state', ui)
monkeypatch.setattr(large, 'live_witness', lambda: read(parked_ui()))
widget = large.SpacePilotDeveloperAdapter(panel, fonts, Mock(), Mock())
widget.set_rect(rl.Rectangle(0, 0, 2160, 1080))
widget.show_event()
widget.gate = mature_gate()
yield widget
widget.hide_event()
def deliver(widget, monkeypatch, index, *, drift=0, duration=.04):
bounds = widget._version_bounds()
pos = MousePos(bounds.x + 20, bounds.y + 10)
now = 10 + index * .2
monkeypatch.setattr(large.time, 'monotonic', lambda: now)
events = [MouseEvent(pos, 0, True, False, True, now)]
if drift:
events.append(MouseEvent(MousePos(pos.x + drift, pos.y), 0, False, False, True, now + .01))
events.append(MouseEvent(pos, 0, False, True, False, now + duration))
monkeypatch.setattr(gui_app, '_mouse_events', events)
widget._process_mouse_events()
def test_big_ui_five_version_value_taps_launch_same_native_game(adapter, monkeypatch):
push = Mock()
monkeypatch.setattr(gui_app, 'push_widget', push)
for i in range(4):
deliver(adapter, monkeypatch, i)
push.assert_not_called()
deliver(adapter, monkeypatch, 4)
push.assert_called_once()
assert isinstance(push.call_args.args[0], native.SpacePilotPage)
assert push.call_args.args[0].gate is not adapter.gate
@pytest.mark.parametrize('cancel', ['drag', 'long', 'page', 'hide', 'unsafe', 'outside', 'scroll'])
def test_big_ui_cancel_resets_deliberate_gesture(adapter, monkeypatch, cancel):
push = Mock()
monkeypatch.setattr(gui_app, 'push_widget', push)
for i in range(4):
deliver(adapter, monkeypatch, i)
if cancel == 'drag':
deliver(adapter, monkeypatch, 4, drift=30)
elif cancel == 'long':
deliver(adapter, monkeypatch, 4, duration=.5)
elif cancel == 'page':
adapter._emit(NS(kind='scroll', direction=1))
elif cancel == 'hide':
adapter.hide_event()
elif cancel == 'unsafe':
monkeypatch.setattr(large, 'live_witness', lambda: None)
adapter._update_state()
elif cancel == 'outside':
monkeypatch.setattr(gui_app, '_mouse_events', [MouseEvent(MousePos(10, 10), 0, True, False, True, 10.9)])
adapter._process_mouse_events()
else:
adapter.set_touch_valid_callback(lambda: False)
monkeypatch.setattr(gui_app, '_mouse_events', [])
adapter._process_mouse_events()
assert adapter.taps.count == 0
push.assert_not_called()
openpilot/starpilot/ui/tests/test_spacepilot_native.py
+71 -63
View File
@@ -33,9 +33,11 @@ def page(monkeypatch):
monkeypatch.setattr(native, 'live_witness', lambda: read(parked_ui()))
monkeypatch.setattr(native.time, 'monotonic', lambda: 10.)
monkeypatch.setattr(native.time, 'monotonic_ns', lambda: 10_000_000_000)
wheel = Mock()
wheel.poll.return_value = False
monkeypatch.setattr(native, 'WheelReader', lambda _: wheel)
game = Mock(stage=native.Stage.PLAYING, reason='', vehicle=Mock(), wheel=Mock(armed=set()))
game.step.return_value = False
game.playable.return_value = True
game.close.side_effect = lambda *_a: setattr(game, 'stage', native.Stage.CLOSED)
monkeypatch.setattr(native, 'GameSession', lambda *_a, **_k: game)
widget = native.SpacePilotPage(mature_gate(), Mock())
widget.set_rect(rl.Rectangle(0, 0, 536, 240))
monkeypatch.setattr(gui_app, 'get_active_widget', lambda: widget)
@@ -54,10 +56,10 @@ def test_native_touch_starts_hold_does_not_repeat_exit_and_cleanup(page):
assert page.flight.vy == 10
page._handle_mouse_event(MouseEvent(MousePos(160, 120), 0, False, True, False, 10.1))
page._handle_mouse_press(MousePos(500, 20))
wheel = page.wheel
game = page.session
page._handle_mouse_event(MouseEvent(MousePos(500, 20), 0, False, True, False, 10.2))
assert page.closed and page.wheel is None
wheel.close.assert_called_once()
assert page.closed and page.session.stage == native.Stage.CLOSED
game.close.assert_called_once()
gui_app.pop_widget.assert_called_once()
assert page._tick not in gui_app._nav_stack_ticks
assert not page.flight.gates
@@ -65,18 +67,13 @@ def test_native_touch_starts_hold_does_not_repeat_exit_and_cleanup(page):
@pytest.mark.parametrize('reason', ['stale', 'hidden', 'backlog', 'no_socket'])
def test_safety_exit_reveals_main_and_never_renders_or_continues_input(page, monkeypatch, reason):
wheel = page.wheel
if reason == 'stale':
monkeypatch.setattr(native, 'live_witness', lambda: None)
elif reason == 'hidden':
if reason == 'hidden':
monkeypatch.setattr(gui_app, 'get_active_widget', lambda: object())
elif reason == 'backlog':
wheel.poll.side_effect = ValueError('backlog')
else:
page.wheel = None
page.session.step.side_effect = lambda: setattr(page.session, 'stage', native.Stage.CLOSED)
page._tick()
page.on_unsafe.assert_called_once()
assert page.closed and page.wheel is None
assert page.closed and page.session.stage == native.Stage.CLOSED
page.renderer = Mock()
page._render(page.rect)
page.renderer.render.assert_not_called()
@@ -85,10 +82,10 @@ def test_safety_exit_reveals_main_and_never_renders_or_continues_input(page, mon
def test_wheel_only_observes_admitted_game_and_drives_same_action(page):
page.wheel.poll.return_value = True
page.session.step.return_value = True
page._tick()
assert page.flight.phase == Phase.FLIGHT
page.wheel.poll.assert_called_once_with(10_000_000_000)
page.session.step.assert_called_once_with()
def test_optional_render_failure_returns_to_native_alerts(page, monkeypatch):
@@ -96,7 +93,7 @@ def test_optional_render_failure_returns_to_native_alerts(page, monkeypatch):
page.renderer.render.side_effect = RuntimeError('GL fixture failure')
monkeypatch.setattr(native.cloudlog, 'exception', Mock())
page._render(page.rect)
assert page.closed and page.wheel is None
assert page.closed and page.session.stage == native.Stage.CLOSED
page.on_unsafe.assert_called_once()
@@ -147,52 +144,6 @@ def test_version_sequence_resets_on_interruption(monkeypatch, interruption):
pushed.assert_not_called()
def car_event(stamp, *, pressed=None, kind='accelCruise'):
event = messaging.new_message('carState', valid=True)
event.logMonoTime = stamp
event.carState.canValid = True
event.carState.standstill = True
event.carState.gearShifter = 'park'
if pressed is not None:
event.carState.buttonEvents = [{'type': kind, 'pressed': pressed}]
return event.as_reader()
def test_nonconflated_reader_retains_short_press_release_and_closes(monkeypatch):
socket = Mock()
subscribe = Mock(return_value=socket)
monkeypatch.setattr(messaging, 'sub_sock', subscribe)
reader = native.WheelReader(1_000_000_000)
subscribe.assert_called_once_with('carState', conflate=False)
messages = iter([car_event(1_010_000_000, pressed=False), car_event(1_020_000_000, pressed=True),
car_event(1_030_000_000, pressed=False), None])
monkeypatch.setattr(messaging, 'recv_one_or_none', lambda _: next(messages))
assert reader.poll(1_040_000_000)
assert 'accelCruise' in reader.input.armed
reader.close()
assert reader.socket is None and not reader.input.armed
@pytest.mark.parametrize('fault', ['stale', 'future', 'invalid', 'drive', 'moving', 'backlog'])
def test_reader_rejects_untrustworthy_queued_input(monkeypatch, fault):
monkeypatch.setattr(messaging, 'sub_sock', Mock())
reader = native.WheelReader(1_000_000_000)
message = car_event(1_100_000_000).as_builder()
if fault == 'future':
message.logMonoTime = 2_000_000_000
elif fault == 'invalid':
message.valid = False
elif fault == 'drive':
message.carState.gearShifter = 'drive'
elif fault == 'moving':
message.carState.vEgo = 2.
messages = iter([message.as_reader(), None])
monkeypatch.setattr(messaging, 'recv_one_or_none', lambda _: message.as_reader() if fault == 'backlog' else next(messages))
with pytest.raises(ValueError):
reader.poll(1_500_000_000 if fault == 'stale' else 1_110_000_000)
reader.close()
def test_five_tap_native_value_region_movement_and_page_reset(monkeypatch):
monkeypatch.setattr(gui_app, 'texture', lambda *_a, **_k: NS(width=64, height=64))
monkeypatch.setattr(gui_app, 'font', lambda *_a: rl.Font())
@@ -256,3 +207,60 @@ def test_developer_installs_version_gesture_only_for_c4_with_exit_owner(monkeypa
monkeypatch.setattr(gui_app, 'font', lambda *_a: rl.Font())
panel = developer.DeveloperLayoutMici(on_spacepilot_exit=Mock() if has_owner else None)
assert sum(isinstance(item, native.VersionButton) for item in panel._scroller.items) == expected
def test_entry_gate_does_not_share_background_settings_reset(page):
gate = mature_gate()
new_page = native.SpacePilotPage(gate, Mock())
gate.reset()
assert new_page.gate.ready
new_page.session.close()
def test_preparation_never_starts_game_and_exit_works(page):
page.session.stage = native.Stage.STOPPING
page._handle_mouse_press(MousePos(160, 120))
assert page.flight.phase == Phase.READY
page._handle_mouse_event(MouseEvent(MousePos(160, 120), 0, False, True, False, 10.1))
page._handle_mouse_press(MousePos(500, 20))
page._handle_mouse_event(MouseEvent(MousePos(500, 20), 0, False, True, False, 10.2))
assert page.closed
def test_controller_start_debounce_and_unsafe_input_isolated(page):
page.capture_controller(native.INCREASE_SPEED, 10_100_000_000)
assert page.flight.phase == Phase.FLIGHT
page.flight.vy = 10
page.capture_controller(native.DECREASE_SPEED, 10_110_000_000)
assert page.flight.vy == 10
page.capture_controller('unrelated.action', 10_400_000_000)
assert page.flight.vy == 10
page.session.playable.return_value = False
page.capture_controller(native.INCREASE_SPEED, 10_600_000_000)
assert page.closed
def test_big_ui_canvas_preserves_aspect_and_exit_target(page):
from openpilot.starpilot.ui.spacepilot.render import canvas_rect
page.set_rect(rl.Rectangle(0, 0, 2160, 1080))
canvas = canvas_rect(page.rect)
assert canvas.width / canvas.height == pytest.approx(1072 / 480)
assert canvas.y > 0
assert page._in_exit(MousePos(canvas.x + 997 * canvas.width / 1072, canvas.y + 25 * canvas.height / 480))
assert not page._in_exit(MousePos(997, 25))
def test_real_nav_stack_launch_survives_background_gesture_reset(monkeypatch, page):
version = Mock()
root = Mock()
monkeypatch.setattr(gui_app, '_nav_stack', [root, version])
monkeypatch.setattr(gui_app, 'get_active_widget', type(gui_app).get_active_widget.__get__(gui_app))
monkeypatch.setattr(gui_app, 'pop_widget', type(gui_app).pop_widget.__get__(gui_app))
game = native.SpacePilotPage(mature_gate(), Mock())
gui_app.push_widget(game)
assert gui_app.get_active_widget() is game
version.set_enabled.assert_called_once_with(False)
game._tick()
assert not game.closed
gui_app.pop_widget()
assert game.closed and game._tick not in gui_app._nav_stack_ticks
openpilot/starpilot/ui/tests/test_spacepilot_session.py
+267
View File
@@ -0,0 +1,267 @@
"""Exclusive game admission, real drive-state requests, and unsafe lifecycle exits."""
from types import SimpleNamespace as NS
from unittest.mock import Mock
import pytest
from openpilot.common.params import Params
from openpilot.starpilot.drive_state.owner import DriveStateOwner
from openpilot.starpilot.ui.spacepilot import session
from openpilot.starpilot.ui.spacepilot.guard import DRIVING_PROCESSES, read_offroad_witness
from openpilot.starpilot.ui.tests.test_spacepilot_state import parked_ui
class Clock:
now = 10_000_000_000
offset = 2_000_000_000
def refresh(self, ui, step=100_000_000):
self.now += step
for name in ui.sm:
ui.sm.valid[name] = ui.sm.alive[name] = True
ui.sm.recv_frame[name] = 10
ui.sm.recv_time[name] = self.now / 1e9
ui.sm.logMonoTime[name] = self.now + (self.offset if name == 'pandaStates' else 0)
class Vehicle:
def __init__(self, cp):
self.cp = cp
self.fresh = True
self.closed = False
self.sample = NS(source_boot_ns=1)
self.samples = ()
def poll(self):
result, self.samples = self.samples, ()
return result
def ready(self, *_):
return self.fresh and not self.closed
def close(self):
self.closed = True
def game_ui(*, ignition=True):
ui = parked_ui()
ui.params = NS(get_bool=lambda _: not ui.started)
ui.sm['deviceState'].thermalStatus = 'ok'
ui.sm['managerState'] = NS(processes=[NS(name=name, running=True) for name in DRIVING_PROCESSES])
ui.sm['carParams'] = ui.CP
ui.sm['pandaStates'][0].safetyModel = 'hyundai'
ui.sm['pandaStates'][0].ignitionLine = ignition
return ui
def offroad(ui):
ui.started = ui.sm['deviceState'].started = False
for process in ui.sm['managerState'].processes:
process.running = False
ui.sm['pandaStates'][0].safetyModel = 'noOutput'
@pytest.fixture
def setup(monkeypatch, tmp_path):
clock = Clock()
ui = game_ui()
clock.refresh(ui, 0)
params = Params(str(tmp_path / 'params'))
owner = DriveStateOwner(params, tmp_path / 'drive-state')
owner.initialize_manager()
monkeypatch.setattr(session.GameSession, '_clocks', lambda _: (clock.now, clock.now + clock.offset))
game = session.GameSession(ui, owner, native_device=True, reader_factory=Vehicle)
game.started_ns = clock.now
yield game, ui, owner, clock
game.close()
def enter(game, ui, clock, *, request=True):
for _ in range(12):
game.step()
clock.refresh(ui)
if request:
assert game.stage == session.Stage.STOPPING
assert game.requested_offroad
offroad(ui)
for _ in range(12):
game.step()
clock.refresh(ui)
assert game.stage == session.Stage.PLAYING, game.reason
def test_actual_owner_request_waits_for_processes_nooutput_then_stays_offroad(setup):
game, ui, owner, clock = setup
initial = owner.snapshot()
for _ in range(12):
assert not game.step()
clock.refresh(ui)
assert owner.snapshot().mode == 'offroad' and owner.snapshot().revision != initial.revision
assert game.stage == session.Stage.STOPPING and not game.playable()
ui.started = ui.sm['deviceState'].started = False
for _ in range(12):
game.step()
clock.refresh(ui)
assert not game.playable()
offroad(ui)
for _ in range(12):
game.step()
clock.refresh(ui)
assert game.playable()
vehicle = game.vehicle
state = owner.snapshot()
game.close()
assert vehicle.closed and game.vehicle is None
assert owner.snapshot() == state and state.mode == 'offroad'
def test_offroad_ignition_off_touch_never_creates_vehicle_reader_or_changes_mode(setup):
game, ui, owner, clock = setup
ui.sm['pandaStates'][0].ignitionLine = False
offroad(ui)
initial = owner.snapshot()
enter(game, ui, clock, request=False)
assert game.vehicle is None and game.playable()
assert owner.snapshot() == initial
assert not game.wheel.armed
def test_offroad_with_ignition_requires_independent_fresh_park(setup):
game, ui, owner, clock = setup
offroad(ui)
game.step()
game.vehicle.fresh = False
for _ in range(20):
clock.refresh(ui)
game.step()
assert not game.playable() and game.stage == session.Stage.VERIFYING
game.vehicle.fresh = True
enter(game, ui, clock, request=False)
assert game.playable()
assert owner.snapshot().mode == 'auto'
@pytest.mark.parametrize('change', ['source', 'receiver', 'future', 'invalid', 'dead'])
@pytest.mark.parametrize('service', ['deviceState', 'pandaStates', 'managerState'])
def test_active_game_exits_for_every_lost_transport(setup, service, change):
game, ui, _, clock = setup
enter(game, ui, clock)
if change == 'source':
ui.sm.logMonoTime[service] -= 2_000_000_000
elif change == 'receiver':
ui.sm.recv_time[service] -= 2
elif change == 'future':
ui.sm.logMonoTime[service] += 2_000_000_000
else:
getattr(ui.sm, 'valid' if change == 'invalid' else 'alive')[service] = False
assert not game.playable()
game.step()
assert game.stage == session.Stage.CLOSED and game.vehicle is None
@pytest.mark.parametrize('change', ['ignition', 'controls', 'safety', 'fault', 'harness', 'process', 'missing_process',
'duplicate_process', 'started', 'raw_stale', 'thermal', 'drive_mode'])
def test_unsafe_exit_and_no_automatic_restore(setup, change):
game, ui, owner, clock = setup
enter(game, ui, clock)
panda = ui.sm['pandaStates'][0]
if change == 'ignition':
panda.ignitionLine = False
elif change == 'controls':
panda.controlsAllowed = True
elif change == 'safety':
panda.safetyModel = 'hyundai'
elif change == 'fault':
panda.heartbeatLost = True
elif change == 'harness':
panda.harnessStatus = 'notConnected'
elif change == 'process':
ui.sm['managerState'].processes[0].running = True
elif change == 'missing_process':
ui.sm['managerState'].processes.pop()
elif change == 'duplicate_process':
ui.sm['managerState'].processes.append(ui.sm['managerState'].processes[0])
elif change == 'started':
ui.started = ui.sm['deviceState'].started = True
elif change == 'thermal':
ui.sm['deviceState'].thermalStatus = 'overheated'
elif change == 'raw_stale':
game.vehicle.fresh = False
else:
owner.request('auto', expected_revision=owner.snapshot().revision, authorized=lambda: True, override_allowed=lambda: True)
expected = owner.snapshot()
assert not game.playable()
game.step()
assert game.stage == session.Stage.CLOSED
assert owner.snapshot() == expected
def test_wheel_requires_post_admission_release_and_does_not_repeat(setup):
game, ui, _, clock = setup
enter(game, ui, clock)
def edge(pressed):
clock.refresh(ui, 140_000_000)
game.vehicle.samples = (NS(source_boot_ns=clock.now + clock.offset, buttons=(('accelCruise', pressed),)),)
return game.step()
assert not edge(True)
assert not edge(False)
assert edge(True)
assert not edge(True)
assert not edge(False)
assert edge(True)
def test_current_vehicle_identity_required_and_forced_flag_alone_never_admits(setup):
game, ui, owner, clock = setup
initial = owner.snapshot()
ui.sm.recv_frame['carParams'] = ui.started_frame
game.step()
assert game.stage == session.Stage.CLOSED and owner.snapshot() == initial
offroad(ui)
assert read_offroad_witness(ui, clock.now, clock.now + clock.offset, native_device=True) is None
assert read_offroad_witness(ui, clock.now, clock.now + clock.offset, native_device=True, preparing=True) is not None
def test_pending_transition_timeout_and_constructor_error_cleanup(setup):
game, ui, owner, clock = setup
game._reader_factory = Mock(side_effect=ValueError('Unsupported decoder'))
game.step()
assert game.stage == session.Stage.CLOSED and owner.snapshot().mode == 'auto'
timed = session.GameSession(ui, owner, native_device=True, reader_factory=Vehicle)
timed.started_ns = clock.now - session.TRANSITION_NS - 1
timed.step()
assert timed.stage == session.Stage.CLOSED and timed.vehicle is None
def test_device_resume_invalidates_session_clock(monkeypatch):
game = session.GameSession(NS(), NS(), native_device=True)
mono, boot = [10_000_000_000], [12_000_000_000]
monkeypatch.setattr(session.time, 'monotonic_ns', lambda: mono[0])
monkeypatch.setattr(session.time, 'clock_gettime_ns', lambda _: boot[0])
assert game._clocks() == (mono[0], boot[0])
boot[0] += 6_000_000
with pytest.raises(ValueError, match='resumed'):
game._clocks()
def test_offroad_witness_accepts_actual_schema_and_refuses_actual_thermal_faults(setup):
from openpilot.cereal import messaging
game, ui, _, clock = setup
offroad(ui)
ui.sm['pandaStates'][0].ignitionLine = False
for name in ('deviceState', 'pandaStates', 'managerState'):
event = messaging.new_message(name, 1 if name == 'pandaStates' else None, valid=True)
if name == 'pandaStates':
event.pandaStates[0] = vars(ui.sm[name][0])
elif name == 'managerState':
event.managerState.processes = [vars(p) for p in ui.sm[name].processes]
else:
event.deviceState = vars(ui.sm[name])
ui.sm[name] = getattr(event.as_reader(), name)
assert read_offroad_witness(ui, clock.now, clock.now + clock.offset, native_device=True) is not None
for thermal in ('overheated', 'critical'):
message = ui.sm['deviceState'].as_builder()
message.thermalStatus = thermal
ui.sm['deviceState'] = message.as_reader()
assert read_offroad_witness(ui, clock.now, clock.now + clock.offset, native_device=True) is None
prebuilt.json
+1 -1
View File
@@ -1 +1 @@
{"protected_inventory_sha256": "60d8463380805b9b19fbeaf7173da556c139c91463a50ba7592df033efb05316", "schema_version": 1}
{"protected_inventory_sha256": "f189d3ad0a84048ebbdb53570e5b703c63b50e1c91f6ea73bd167b46450d4b6e", "schema_version": 1}