mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-28 00:53:45 +08:00
firestar5683
106 lines
5.7 KiB
Python
106 lines
5.7 KiB
Python
"""Startup admission for saved optional driving features.
|
|
|
|
This decides which optional owners exist for a drive. It never grants control;
|
|
each owner still verifies current vehicle, source, clock, and axis authority.
|
|
"""
|
|
|
|
from collections.abc import Mapping
|
|
from dataclasses import replace
|
|
|
|
from opendbc.car.gm.feature_capabilities import longitudinal_supported as gm_long_supported
|
|
|
|
from openpilot.starpilot.conditional_mode.preferences import PreferenceError, decode_preferences
|
|
from openpilot.starpilot.aol.intent import independent_axis_requested, read_settings as read_aol_settings
|
|
from openpilot.starpilot.aol.vehicle import policy_for as aol_policy_for
|
|
from openpilot.starpilot.feature_capabilities import display_supported, longitudinal_supported
|
|
from openpilot.starpilot.longitudinal.profile_runtime import read_settings as read_profile_settings, selected_profiles_requested, read_global_powertrain_preset
|
|
from openpilot.starpilot.longitudinal.profile_preferences import read_document_value
|
|
from openpilot.starpilot.saved_source import read_saved
|
|
from openpilot.starpilot.speed_limits.runtime_settings import Settings as SlcSettings, read_params as read_slc_settings
|
|
from openpilot.starpilot.speed_limits.selection import Source
|
|
|
|
|
|
def _read(params, key: str, limit: int = 4096) -> tuple[bytes | None, bool]:
|
|
try:
|
|
return read_saved(params, key, limit)
|
|
except (OSError, TypeError, ValueError):
|
|
return None, False
|
|
|
|
|
|
def requested(params, feature: str, *, include_auxiliary: bool = False, saved_slc: SlcSettings | None = None) -> bool:
|
|
"""Read one feature request, including the readable factory CEM default."""
|
|
if feature == 'conditional':
|
|
raw, document_readable = _read(params, 'ConditionalModeConfig')
|
|
safe, safe_readable = _read(params, 'SafeMode', 8)
|
|
if not document_readable or not safe_readable or safe not in (None, b'0'):
|
|
return False
|
|
if raw is None:
|
|
return True # Readable absence selects the factory CEM choice.
|
|
try:
|
|
decode_preferences(raw)
|
|
return True # Stock still owns its optional wheel/session lifecycle.
|
|
except PreferenceError:
|
|
return False
|
|
if feature == 'curve':
|
|
return _read(params, 'CurveSpeedController', 8) == (b'1', True)
|
|
if feature == 'aol':
|
|
master, readable = _read(params, 'AlwaysOnLateral', 8)
|
|
if not readable or master not in (None, b'0', b'1'):
|
|
return False
|
|
return independent_axis_requested(read_aol_settings(params), include_auxiliary=include_auxiliary)
|
|
if feature == 'profile':
|
|
document = read_document_value(params)
|
|
return (read_profile_settings(params) is not None or
|
|
(document.valid and isinstance(document.value, dict) and
|
|
selected_profiles_requested(document.value)))
|
|
if feature in ('slc', 'vision'):
|
|
settings = read_slc_settings(params) if saved_slc is None else saved_slc
|
|
return (settings.enabled or settings.display) if feature == 'slc' else (
|
|
(settings.enabled or settings.display) and Source.VISION in settings.selection.slots)
|
|
raise ValueError('unknown optional feature')
|
|
|
|
|
|
def enabled(params, cp, feature: str, environment: Mapping[str, str], *, saved_slc: SlcSettings | None = None) -> bool:
|
|
"""Preserve explicit replay opt-ins and each vehicle capability."""
|
|
flags = {'conditional': 'CONDITIONAL_MODE_REPLAY_RUNTIME', 'curve': 'CURVE_REPLAY_RUNTIME',
|
|
'aol': 'AOL_REPLAY_RUNTIME',
|
|
'slc': 'SLC_REPLAY_RUNTIME', 'vision': 'SLC_VISION_DEVELOPMENT',
|
|
'profile': 'LONG_PLANNER_REPLAY_RUNTIME'}
|
|
if feature not in flags:
|
|
raise ValueError('unknown optional feature')
|
|
if feature == 'aol':
|
|
policy = aol_policy_for(cp)
|
|
return bool(policy.runtime_supported and
|
|
(environment.get(flags[feature]) == '1' or
|
|
(policy.normal_runtime_supported and (policy.full_axis_runtime_required or policy.physical_stalk_owner or requested(params, feature,
|
|
include_auxiliary=policy.intent_supported and not policy.explicit_latch)))))
|
|
if feature == 'vision' and environment.get('SLC_REPLAY_RUNTIME') == '1' and environment.get(flags[feature]) == '1':
|
|
return True
|
|
if feature != 'vision' and environment.get(flags[feature]) == '1':
|
|
return True
|
|
capable = display_supported(cp) if feature in ('slc', 'vision') else longitudinal_supported(cp)
|
|
powertrain_requested = (feature == 'profile' and cp is not None and gm_long_supported(cp) and
|
|
read_document_value(params).valid and read_global_powertrain_preset(params, cp) is not None)
|
|
return bool(capable and (requested(params, feature, saved_slc=saved_slc) or powertrain_requested))
|
|
|
|
|
|
def slc_transport_capable(cp, environment: Mapping[str, str]) -> bool:
|
|
"""Reserve optional transport without admitting a saved request or control."""
|
|
return bool(environment.get('SLC_REPLAY_RUNTIME') == '1' or
|
|
display_supported(cp))
|
|
|
|
|
|
def slc_runtime_settings(params, cp, environment: Mapping[str, str], *, saved_slc: SlcSettings | None = None):
|
|
"""Stock display startup cannot inherit a saved speed-control request."""
|
|
settings = read_slc_settings(params) if saved_slc is None else saved_slc
|
|
if environment.get('SLC_REPLAY_RUNTIME') != '1' and not longitudinal_supported(cp):
|
|
return replace(settings, enabled=False, display=settings.display or settings.enabled, acceptance=replace(settings.acceptance, display_only=True))
|
|
return settings
|
|
|
|
|
|
def vision_control_enabled(params, cp, *, saved_slc: SlcSettings | None = None) -> bool:
|
|
"""Source selection never turns stock-cruise or display-only sessions into control."""
|
|
settings = read_slc_settings(params) if saved_slc is None else saved_slc
|
|
return bool(longitudinal_supported(cp) and
|
|
settings.enabled and Source.VISION in settings.selection.slots)
|