From 347397ad384441aec32804d626e0002ac2122c4e Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Mon, 26 Oct 2026 00:01:36 -0500 Subject: [PATCH] Add exclusive parked SpacePilot mode and native Big UI 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. --- openpilot/selfdrive/ui/ui.py | 6 +- openpilot/starpilot/controllers/owner.py | 14 +- openpilot/starpilot/controllers/runtime.py | 9 +- .../controllers/tests/test_capture.py | 55 ++++ openpilot/starpilot/parked_vehicle.py | 111 ++++++++ .../starpilot/tests/test_parked_vehicle.py | 138 +++++++++ openpilot/starpilot/ui/runtime_app.py | 11 +- openpilot/starpilot/ui/spacepilot/README.md | 89 +++--- openpilot/starpilot/ui/spacepilot/guard.py | 58 +++- openpilot/starpilot/ui/spacepilot/large.py | 113 ++++++++ openpilot/starpilot/ui/spacepilot/native.py | 134 ++++----- openpilot/starpilot/ui/spacepilot/render.py | 40 ++- openpilot/starpilot/ui/spacepilot/session.py | 166 +++++++++++ .../ui/tests/test_spacepilot_large.py | 83 ++++++ .../ui/tests/test_spacepilot_native.py | 134 ++++----- .../ui/tests/test_spacepilot_session.py | 267 ++++++++++++++++++ prebuilt.json | 2 +- 17 files changed, 1245 insertions(+), 185 deletions(-) create mode 100644 openpilot/starpilot/controllers/tests/test_capture.py create mode 100644 openpilot/starpilot/parked_vehicle.py create mode 100644 openpilot/starpilot/tests/test_parked_vehicle.py create mode 100644 openpilot/starpilot/ui/spacepilot/large.py create mode 100644 openpilot/starpilot/ui/spacepilot/session.py create mode 100644 openpilot/starpilot/ui/tests/test_spacepilot_large.py create mode 100644 openpilot/starpilot/ui/tests/test_spacepilot_session.py diff --git a/openpilot/selfdrive/ui/ui.py b/openpilot/selfdrive/ui/ui.py index d49efde23a..80a7b8e241 100755 --- a/openpilot/selfdrive/ui/ui.py +++ b/openpilot/selfdrive/ui/ui.py @@ -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: diff --git a/openpilot/starpilot/controllers/owner.py b/openpilot/starpilot/controllers/owner.py index f9c5a99804..9b4f57fd8c 100644 --- a/openpilot/starpilot/controllers/owner.py +++ b/openpilot/starpilot/controllers/owner.py @@ -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 diff --git a/openpilot/starpilot/controllers/runtime.py b/openpilot/starpilot/controllers/runtime.py index 8418fa7888..7969ad43c1 100644 --- a/openpilot/starpilot/controllers/runtime.py +++ b/openpilot/starpilot/controllers/runtime.py @@ -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') diff --git a/openpilot/starpilot/controllers/tests/test_capture.py b/openpilot/starpilot/controllers/tests/test_capture.py new file mode 100644 index 0000000000..be0d6874a0 --- /dev/null +++ b/openpilot/starpilot/controllers/tests/test_capture.py @@ -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() diff --git a/openpilot/starpilot/parked_vehicle.py b/openpilot/starpilot/parked_vehicle.py new file mode 100644 index 0000000000..28d489691f --- /dev/null +++ b/openpilot/starpilot/parked_vehicle.py @@ -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() diff --git a/openpilot/starpilot/tests/test_parked_vehicle.py b/openpilot/starpilot/tests/test_parked_vehicle.py new file mode 100644 index 0000000000..73153992e0 --- /dev/null +++ b/openpilot/starpilot/tests/test_parked_vehicle.py @@ -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')) diff --git a/openpilot/starpilot/ui/runtime_app.py b/openpilot/starpilot/ui/runtime_app.py index ba8b278022..3eeee95b4f 100644 --- a/openpilot/starpilot/ui/runtime_app.py +++ b/openpilot/starpilot/ui/runtime_app.py @@ -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( diff --git a/openpilot/starpilot/ui/spacepilot/README.md b/openpilot/starpilot/ui/spacepilot/README.md index e63fc3290f..1bf5495b22 100644 --- a/openpilot/starpilot/ui/spacepilot/README.md +++ b/openpilot/starpilot/ui/spacepilot/README.md @@ -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. diff --git a/openpilot/starpilot/ui/spacepilot/guard.py b/openpilot/starpilot/ui/spacepilot/guard.py index 7fb24982da..63eee17dc0 100644 --- a/openpilot/starpilot/ui/spacepilot/guard.py +++ b/openpilot/starpilot/ui/spacepilot/guard.py @@ -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 diff --git a/openpilot/starpilot/ui/spacepilot/large.py b/openpilot/starpilot/ui/spacepilot/large.py new file mode 100644 index 0000000000..8ae397512e --- /dev/null +++ b/openpilot/starpilot/ui/spacepilot/large.py @@ -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) diff --git a/openpilot/starpilot/ui/spacepilot/native.py b/openpilot/starpilot/ui/spacepilot/native.py index d0d32d047b..71cab82b48 100644 --- a/openpilot/starpilot/ui/spacepilot/native.py +++ b/openpilot/starpilot/ui/spacepilot/native.py @@ -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) diff --git a/openpilot/starpilot/ui/spacepilot/render.py b/openpilot/starpilot/ui/spacepilot/render.py index cd6a159826..a18702820b 100644 --- a/openpilot/starpilot/ui/spacepilot/render.py +++ b/openpilot/starpilot/ui/spacepilot/render.py @@ -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() diff --git a/openpilot/starpilot/ui/spacepilot/session.py b/openpilot/starpilot/ui/spacepilot/session.py new file mode 100644 index 0000000000..10002cea49 --- /dev/null +++ b/openpilot/starpilot/ui/spacepilot/session.py @@ -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. diff --git a/openpilot/starpilot/ui/tests/test_spacepilot_large.py b/openpilot/starpilot/ui/tests/test_spacepilot_large.py new file mode 100644 index 0000000000..e41af57917 --- /dev/null +++ b/openpilot/starpilot/ui/tests/test_spacepilot_large.py @@ -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() diff --git a/openpilot/starpilot/ui/tests/test_spacepilot_native.py b/openpilot/starpilot/ui/tests/test_spacepilot_native.py index 4ce48e47a0..a0c717ff1c 100644 --- a/openpilot/starpilot/ui/tests/test_spacepilot_native.py +++ b/openpilot/starpilot/ui/tests/test_spacepilot_native.py @@ -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 diff --git a/openpilot/starpilot/ui/tests/test_spacepilot_session.py b/openpilot/starpilot/ui/tests/test_spacepilot_session.py new file mode 100644 index 0000000000..4d624c9290 --- /dev/null +++ b/openpilot/starpilot/ui/tests/test_spacepilot_session.py @@ -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 diff --git a/prebuilt.json b/prebuilt.json index 3c33787328..6a5d611fe7 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "60d8463380805b9b19fbeaf7173da556c139c91463a50ba7592df033efb05316", "schema_version": 1} +{"protected_inventory_sha256": "f189d3ad0a84048ebbdb53570e5b703c63b50e1c91f6ea73bd167b46450d4b6e", "schema_version": 1}