Drop legacy SLC offsets for simpler setup

Use one saved schedule with editable zero defaults. Preserve current schedules and round displayed offsets and edits to whole units without conversion drift.
This commit is contained in:
firestarsdog
2026-10-31 00:24:31 -04:00
parent 2fd0741f56
commit 7bce4121e5
28 changed files with 322 additions and 465 deletions
-7
View File
@@ -293,13 +293,6 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
{"DriveStateRequest", {CLEAR_ON_MANAGER_START | DONT_LOG, JSON}},
{"OnroadCycleRequested", {CLEAR_ON_MANAGER_START, BOOL}},
{"OpenpilotEnabledToggle", {PERSISTENT, BOOL, "1"}},
{"Offset1", {PERSISTENT, FLOAT, "0.0"}},
{"Offset2", {PERSISTENT, FLOAT, "0.0"}},
{"Offset3", {PERSISTENT, FLOAT, "0.0"}},
{"Offset4", {PERSISTENT, FLOAT, "0.0"}},
{"Offset5", {PERSISTENT, FLOAT, "0.0"}},
{"Offset6", {PERSISTENT, FLOAT, "0.0"}},
{"Offset7", {PERSISTENT, FLOAT, "0.0"}},
{"PandaHeartbeatLost", {CLEAR_ON_MANAGER_START | CLEAR_ON_OFFROAD_TRANSITION, BOOL}},
{"PairingEmail", {PERSISTENT, STRING}},
{"ConnectProvider", {PERSISTENT, STRING, "comma"}},
@@ -39,7 +39,7 @@ class TestConditionalRuntimeLoop(unittest.TestCase):
params.put_bool('SpeedLimitController', True, block=True)
params.put_bool('SLCConfirmation', True, block=True)
params.put('SLCPriority1', 'Dashboard', block=True)
params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (0.,) * 7)), block=True)
params.put('SLCOffsetSchedule', od.to_value(od.default_document(False)), block=True)
owner = plannerd.SavedSlcLifecycle(params, cp, {})
runtime, _ = owner.refresh(2_000_000_000)
sm['carState'].vCruise = sm['carState'].vCruiseCluster = 60.
@@ -77,7 +77,7 @@ class TestConditionalRuntimeLoop(unittest.TestCase):
params.put_bool('SpeedLimitController', True, block=True)
params.put_bool('SLCConfirmation', confirmation, block=True)
params.put('SLCPriority1', 'Dashboard', block=True)
params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (0.,) * 7)), block=True)
params.put('SLCOffsetSchedule', od.to_value(od.default_document(False)), block=True)
owner = plannerd.SavedSlcLifecycle(params, cp, {})
original, _ = owner.refresh(now)
sm['slcDashboardObservation'].speedMps = 65 * CV.MPH_TO_MS
@@ -5,7 +5,6 @@ from importlib.resources import files
from openpilot.starpilot.saved_document import commit_exact
from openpilot.starpilot.saved_source import read_saved
from openpilot.starpilot.speed_limits import offset_document as od
from openpilot.starpilot.ui.feature_settings_state import FeatureRow, FeatureSettingsState
from openpilot.starpilot.ui.slc_offset_feature import SlcOffsetOwner
@@ -32,13 +31,6 @@ class LocalizationSettingsOwner:
raw, readable = read_saved(self.params, key, 8)
valid = readable and raw in (None, b"0", b"1")
allowed = self.authorized() if key == "IsMetric" else self._wheel_side_allowed()
if key == "IsMetric":
saved = self.units.snapshot()
custom_offsets = isinstance(saved.document, od.OffsetDocument) and any(saved.document.offsets_mps)
preserving = saved.raw_document is None and (custom_offsets or saved.raw_control not in (None, b"0"))
if preserving and not self.parked():
allowed = False
reason = "Park once to preserve your saved speed-limit offsets before switching units."
if key == "IsRhdDetected" and not allowed:
reason = "Park the vehicle and close driver camera preview before changing the steering-wheel side."
rows.append(FeatureRow(key, label, ("On" if raw == b"1" else "Off") if valid else "Invalid saved value",
@@ -4,6 +4,7 @@ Rows and their availability still come from the canonical feature owners.
This presentation catalog supplies the purpose before any genuine prerequisite.
"""
from openpilot.starpilot.ui.long_profile_feature import CATEGORY_HELP, VALUE_HELP
from openpilot.starpilot.ui.feature_settings_state import SLC_OFFSET_KEYS
DESCRIPTIONS = {
'DisengageOnAccelerator': 'When enabled, pressing the accelerator pedal will disengage openpilot.',
@@ -66,7 +67,6 @@ DESCRIPTIONS = {
'SLCAdvanceSeconds': 'Offers mapped limits early using road distance and current speed. Existing adoption and confirmation settings still apply.',
'SLCPriority1': 'Chooses the first source checked for a fresh speed limit.',
'SLCPriority2': 'Chooses the source used when the primary source has no fresh speed limit.',
'slc_adopt': 'Keeps the current speed-limit offset schedule fixed when display units change.',
'slc_reset': 'Resets saved speed-limit offsets while retaining valid configured speed ranges.',
'reset_driver_monitoring': 'Clear saved wheel-side detection. Automatic detection restarts on the next drive.',
'AlwaysAllowUploads': 'Allows queued full logs and videos to upload over metered networks. This uses mobile data.',
@@ -216,7 +216,7 @@ def purpose(row):
suffix = next((name for name in VALUE_HELP if key.endswith(name)), None)
value = VALUE_HELP.get(suffix)
return value + ' Blends toward Relaxed as speed rises in Traffic mode.' if value and key.startswith('Traffic') else value
if key in tuple('Offset'+str(index) for index in range(1, 8)):
if key in SLC_OFFSET_KEYS:
return 'Adds or subtracts this offset from the accepted speed limit within the displayed speed range.'
if key.startswith('long_repair:'):
return 'Restores the default value of this invalid longitudinal setting.'
+1 -3
View File
@@ -148,7 +148,7 @@ def _onroad_preference(page: str, key: str) -> bool:
# multi-key reset/repair operations need the offroad transaction boundary;
# their owners independently recheck that requirement at commit time.
return bool(page not in ("developer", "power") and key and key != SETUP_ACTION and key not in (
"IsRhdDetected", "AlphaLongitudinalEnabled", "reset_profiles", "curve_reset", "slc_adopt", "slc_reset",
"IsRhdDetected", "AlphaLongitudinalEnabled", "reset_profiles", "curve_reset", "slc_reset",
"conditional:reset", "conditional:manual_reset",
) and not is_long_confirm_action(key) and not key.startswith("torque_repair:"))
@@ -445,8 +445,6 @@ def _question(row: FeatureRow, request: FeatureSettingsRequest) -> str:
return "Save these source-image crop positions? Confirm alignment on the device before using the preview."
if key.startswith(PIP_FORMAT_PREFIX):
return f"{row.label}? This replaces the saved crop; check alignment on the device."
if key == "slc_adopt":
return "Keep these saved offsets and speed ranges when units change? Prior saved values remain but become inactive."
if key == "slc_reset":
return f"Reset speed-limit offsets to zero? {row.value}. Saved control may resume if its switch is on."
if key == "curve_reset":
@@ -54,13 +54,13 @@
"LoudBlindspotAlert",
"LoudBlindspotAlertWhenDisengaged",
"NostalgiaMode",
"Offset1",
"Offset2",
"Offset3",
"Offset4",
"Offset5",
"Offset6",
"Offset7",
"slc:offset:1",
"slc:offset:2",
"slc:offset:3",
"slc:offset:4",
"slc:offset:5",
"slc:offset:6",
"slc:offset:7",
"PIPPreviewEnabled",
"PIPPreviewInvert",
"PIPPreviewShowOnBSM",
@@ -346,7 +346,6 @@
"sentry:reset",
"sentry:sensitivity",
"sentry:warning",
"slc_adopt",
"slc_reset",
"sounds:auto:DisengageVolume",
"sounds:auto:EngageVolume",
@@ -99,7 +99,7 @@ def test_all_settings_match_ignition_off_in_forced_offroad(powered, vehicle, req
assert rows(page) == ordinary, page
assert next(row for row in rows('aol') if row[0] == 'Enable Always On Lateral')[1]
page = gateway.page('slc', 'session', b'generation')
index = next(i for i, row in enumerate(page['rows']) if row['label'] == 'Adopt fixed offsets')
intent = gateway.preview(page['view'], index, 0, 'session', b'generation')
index = next(i for i, row in enumerate(page['rows']) if row['label'].endswith('offset'))
intent = gateway.preview(page['view'], index, 0, 'session', b'generation', value=3)
assert gateway.confirm(intent['intent'], 'session', b'generation')
assert next(row for row in rows('slc') if row[0] == 'Speed Limit Controller')[1]
@@ -38,16 +38,15 @@ def test_real_gateway_units_and_wheel_side_without_a_vehicle(tmp_path):
assert params.get_bool("IsRhdDetected")
def test_localization_preserves_offsets_and_legacy_values(tmp_path):
def test_localization_preserves_current_offsets(tmp_path):
params = Params(str(tmp_path))
params.put_bool("IsMetric", False, block=True)
params.put("Offset1", 5.0, block=True)
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (5 * od.MPH_TO_MPS, 0, 0, 0, 0, 0, 0))
params.put(DOCUMENT_KEY, od.to_value(document), block=True)
params.put_bool("SpeedLimitController", False, block=True)
owner = LocalizationSettingsOwner(params, lambda: True, lambda: True)
assert owner.apply(FeatureSettingsRequest("IsMetric", b"0", "On"))
document = od.decode(Path(params.get_param_path(DOCUMENT_KEY)).read_bytes())
assert document == od.adopt_legacy(False, (5.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0))
assert Path(params.get_param_path("Offset1")).read_bytes() == b"5.0"
assert od.decode(Path(params.get_param_path(DOCUMENT_KEY)).read_bytes()) == document
assert owner.apply(FeatureSettingsRequest("IsMetric", b"1", "Off"))
assert od.decode(Path(params.get_param_path(DOCUMENT_KEY)).read_bytes()) == document
@@ -86,14 +85,16 @@ def test_units_without_offsets_work_onroad_but_wheel_side_waits(tmp_path):
assert not Path(params.get_param_path(DOCUMENT_KEY)).exists()
def test_legacy_offsets_explain_the_one_time_park_requirement(tmp_path):
def test_retired_offsets_do_not_require_parking_or_migration_to_change_units(tmp_path):
params = Params(str(tmp_path))
params.put("Offset1", 5.0, block=True)
Path(params.get_param_path("Offset1")).write_bytes(b"5.0")
owner = LocalizationSettingsOwner(params, lambda: True, lambda: False)
row = owner.snapshot().rows[0]
assert not row.available and "Park once" in row.reason
assert not owner.apply(FeatureSettingsRequest("IsMetric", None, "On"))
assert not params.get_bool("IsMetric")
assert row.available
assert owner.apply(FeatureSettingsRequest("IsMetric", None, "On"))
assert params.get_bool("IsMetric")
assert not Path(params.get_param_path(DOCUMENT_KEY)).exists()
assert Path(params.get_param_path("Offset1")).read_bytes() == b"5.0"
@pytest.mark.parametrize("name,code", [("English", "en"), ("日本語", "ja"), ("한국어", "ko"),
@@ -78,9 +78,9 @@ class TestOffroadConfigurationPages(unittest.TestCase):
self.assertFalse(context.cp.openpilotLongitudinalControl)
self.assertTrue(context.configuration_longitudinal)
page = self.page('slc')
index = next(i for i, row in enumerate(page['rows']) if row['label'] == 'Adopt fixed offsets')
index = next(i for i, row in enumerate(page['rows']) if row['label'].endswith('offset'))
self.assertTrue(page['rows'][index]['available'])
intent = self.gateway.preview(page['view'], index, 0, 'session', b'generation')
intent = self.gateway.preview(page['view'], index, 0, 'session', b'generation', value=3)
self.assertTrue(self.gateway.confirm(intent['intent'], 'session', b'generation'))
self.assertTrue(next(row for row in self.page('slc')['rows'] if row['label'] == 'Speed Limit Controller')['available'])
self.assertFalse(self.params.get_bool('AlphaLongitudinalEnabled'))
@@ -97,8 +97,8 @@ class TestOffroadConfigurationPages(unittest.TestCase):
def test_saved_and_manual_contexts_expose_expected_core_settings(self):
offsets = self.page('slc')
index = next(i for i, row in enumerate(offsets['rows']) if row['label'] == 'Adopt fixed offsets')
intent = self.gateway.preview(offsets['view'], index, 0, 'session', b'generation')
index = next(i for i, row in enumerate(offsets['rows']) if row['label'].endswith('offset'))
intent = self.gateway.preview(offsets['view'], index, 0, 'session', b'generation', value=3)
self.assertTrue(self.gateway.confirm(intent['intent'], 'session', b'generation'))
expected = {
'aol': ('Enable Always On Lateral', 'Brake pause below'),
@@ -137,9 +137,9 @@ class SettingsGatewayTest(unittest.TestCase):
cp.carFingerprint = "TOYOTA_HIGHLANDER_TSS2"
cp.pcmCruise = True
page = self.page("slc")
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Adopt fixed offsets")
index = next(i for i, row in enumerate(page["rows"]) if row["label"].endswith("offset"))
self.assertTrue(page["rows"][index]["available"])
intent = self.gateway.preview(page["view"], index, 0, self.session, self.generation)
intent = self.gateway.preview(page["view"], index, 0, self.session, self.generation, value=3)
self.assertTrue(self.gateway.confirm(intent["intent"], self.session, self.generation))
row = next(row for row in self.page("slc")["rows"] if row["label"] == "Speed Limit Controller")
self.assertTrue(row["available"])
@@ -219,7 +219,7 @@ class SettingsGatewayTest(unittest.TestCase):
def test_direct_offset_slider_keeps_si_units_and_range(self):
from openpilot.starpilot.speed_limits import offset_document as od
self.params.put("SLCOffsetSchedule", od.to_value(od.adopt_legacy(False, (0.0,) * 7)), block=True)
self.params.put("SLCOffsetSchedule", od.to_value(od.default_document(False)), block=True)
page = self.page("slc")
index = next(i for i, row in enumerate(page["rows"]) if row["label"].endswith("offset"))
self.assertEqual((page["rows"][index]["minimum"], page["rows"][index]["maximum"]), (-25, 99))
@@ -791,7 +791,7 @@ class SettingsGatewayTest(unittest.TestCase):
self.assertFalse(self.gateway.confirm(intent["intent"], self.session, self.generation))
self.assertFalse(self.params.get_bool("LaneCentering"))
def test_slc_adoption_long_preset_and_lane_change_use_owner(self):
def test_slc_first_edit_long_preset_and_lane_change_use_owner(self):
from opendbc.car.car_helpers import interfaces
from opendbc.car.toyota.values import CAR as TOYOTA
@@ -800,10 +800,13 @@ class SettingsGatewayTest(unittest.TestCase):
cp.pcmCruise = False
self.context.value = replace(self.context.value, cp=cp)
slc = self.page("slc")
adopt = next(i for i, row in enumerate(slc["rows"]) if row["label"] == "Adopt fixed offsets")
intent = self.gateway.preview(slc["view"], adopt, 0, self.session, self.generation)
self.assertFalse(any(row["label"] == "Adopt fixed offsets" for row in slc["rows"]))
offset = next(i for i, row in enumerate(slc["rows"]) if row["label"].endswith("offset"))
self.assertEqual(slc["rows"][offset]["value"], "0")
intent = self.gateway.preview(slc["view"], offset, 0, self.session, self.generation, value=3)
self.assertTrue(self.gateway.confirm(intent["intent"], self.session, self.generation))
assert read_saved(self.params, "SLCOffsetSchedule", 4096)[0] is not None
self.assertAlmostEqual(self.params.get("SLCOffsetSchedule")["offsets_mps"][0], 3 * .44704)
page = self.page("aggressive/acceleration")
preset = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Preset")
intent = self.gateway.preview(page["view"], preset, 1, self.session, self.generation)
@@ -0,0 +1,40 @@
"""Converted offsets stay readable and editable through the Galaxy settings API."""
from types import SimpleNamespace
from typing import cast
import pytest
from opendbc.car.toyota.interface import CarInterface
from opendbc.car.toyota.values import CAR
from openpilot.common.params import Params
from openpilot.starpilot.galaxy.settings import AuthorityContext, ContextSource, SettingsChanged, SettingsGateway
from openpilot.starpilot.speed_limits import offset_document as od
@pytest.mark.parametrize("metric", (False, True))
def test_converted_offset_slider_uses_whole_steps_in_both_units(tmp_path, metric):
params = Params(str(tmp_path))
params.put_bool("IsMetric", metric, block=True)
cp = CarInterface.get_non_essential_params(CAR.TOYOTA_COROLLA_TSS2)
cp.openpilotLongitudinalControl, cp.pcmCruise = True, False
cp_raw = cp.to_bytes()
context = SimpleNamespace(sample=lambda: AuthorityContext(True, cp, cp_raw, metric=metric))
gateway = SettingsGateway(params, cast(ContextSource, context))
try:
source = 3 * od.MPH_TO_MPS if metric else 5 * od.KPH_TO_MPS
document = od.OffsetDocument(od.default_document(not metric).bounds_mps, (source,) * 7)
params.put("SLCOffsetSchedule", od.to_value(document), block=True)
page = gateway.page("slc", "session", b"generation")
index = next(i for i, row in enumerate(page["rows"]) if row["label"].endswith("offset"))
expected = 5 if metric else 3
assert page["rows"][index]["value"] == str(expected)
with pytest.raises(SettingsChanged):
gateway.preview(page["view"], index, 0, "session", b"generation", value=4.83)
intent = gateway.preview(page["view"], index, 0, "session", b"generation", value=expected + 1)
assert gateway.confirm(intent["intent"], "session", b"generation")
factor = od.KPH_TO_MPS if metric else od.MPH_TO_MPS
assert params.get("SLCOffsetSchedule")["offsets_mps"][0] == pytest.approx((expected + 1) * factor)
assert gateway.page("slc", "session", b"generation")["rows"][index]["value"] == str(expected + 1)
finally:
gateway.close()
@@ -125,6 +125,15 @@ saved bytes for older versions; current code neither reads nor writes it.
Missing, 0, 1, 2 and malformed old values all use the same behavior above.
Old Quick Select assignments remain saved but unavailable.
Offsets use only the saved `SLCOffsetSchedule`, with fixed physical speed ranges
and adjustments. Without a saved schedule, all seven ranges start at zero and
can be edited immediately in Galaxy, Large UI, or Compact UI. Retired
`Offset1`–`Offset7` values are ignored; users with only those values must configure
their offsets again. Existing schedules are preserved, including across display
unit changes. There is no separate offset-adoption step.
Converted offsets display as whole mph or km/h, and edits use whole units.
Unit switches retain the saved precision so repeated conversions do not drift.
Offset-only edits preserve accepted and
rejected state and renew action identities. A changed offset can update the set
speed of the accepted current limit; source loss defers this command until that
@@ -1,4 +1,4 @@
"""Strict SI authority for adopted SLC offsets; legacy source bytes stay untouched."""
"""Speed-limit offset ranges and adjustments stored in fixed SI units."""
from __future__ import annotations
@@ -28,7 +28,7 @@ class OffsetDocument:
@dataclass(frozen=True)
class NeedsReview:
"""A unit change with corrupt saved control settings must not restore control."""
"""Saved offsets require an explicit reset before control can resume."""
NEEDS_REVIEW = NeedsReview()
@@ -40,12 +40,10 @@ def _finite(value: object) -> float:
return float(value)
def adopt_legacy(metric: bool, offsets: tuple[object, ...]) -> OffsetDocument:
if type(metric) is not bool or len(offsets) != 7:
raise ValueError('invalid legacy SLC offsets or units')
conversion = KPH_TO_MPS if metric else MPH_TO_MPS
bounds = METRIC_BOUNDS if metric else IMPERIAL_BOUNDS
return OffsetDocument(bounds, tuple(_finite(_finite(value) * conversion) for value in offsets))
def default_document(metric: bool) -> OffsetDocument:
if type(metric) is not bool:
raise ValueError('invalid SLC offset units')
return OffsetDocument(METRIC_BOUNDS if metric else IMPERIAL_BOUNDS, (0.0,) * 7)
def validate(value: object) -> OffsetDocument | NeedsReview:
@@ -1,6 +1,5 @@
"""Reviewed saved-setting mapping for the opt-in system-longitudinal SLC runtime."""
import math
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
@@ -37,13 +36,6 @@ def _bool(values: Mapping[str, object], key: str, default: bool = False) -> bool
return value
def _number(values: Mapping[str, object], key: str, default: float = 0.0) -> float:
value = values.get(key, default)
if (type(value) is not int and type(value) is not float) or not math.isfinite(value):
raise ValueError(f"{key} must be finite")
return float(value)
def _disabled(reason: str) -> Settings:
selection = sel.SelectionPolicy(sel.SelectionMode.ORDERED, (sel.Source.DASHBOARD, sel.Source.MAP), False)
return Settings(False, False, selection, acc.Policy(display_only=True),
@@ -55,10 +47,10 @@ def parse(values: Mapping[str, object]) -> Settings:
try:
control = _bool(values, "SpeedLimitController")
display = control or _bool(values, "ShowSpeedLimits", True)
document = od.validate(values['SLCOffsetSchedule']) if 'SLCOffsetSchedule' in values else None
if document is od.NEEDS_REVIEW:
document = (od.validate(values['SLCOffsetSchedule']) if 'SLCOffsetSchedule' in values else
od.default_document(_bool(values, 'IsMetric')))
if not isinstance(document, od.OffsetDocument):
raise ValueError('saved SLC offsets need review')
metric = _bool(values, "IsMetric") if document is None else False # presentation units do not own adopted control
first = values.get("SLCPriority1", "Vision")
second = values.get("SLCPriority2", "Dashboard")
if type(first) is not str or type(second) is not str:
@@ -76,8 +68,7 @@ def parse(values: Mapping[str, object]) -> Settings:
advance = 0 # An invalid optional timer cannot enable early adoption.
policy = acc.Policy(confirm_lower=confirmation, confirm_higher=confirmation,
display_only=not control)
offsets = (document if isinstance(document, od.OffsetDocument) else
od.adopt_legacy(metric, tuple(_number(values, f"Offset{i + 1}") for i in range(7)))).schedule()
offsets = document.schedule()
# SLCOverride remains a validated legacy preference; only set-speed changes override SLC.
return Settings(control, display, sel.SelectionPolicy(mode, slots, False), policy, offsets,
requested=control, advance_seconds=advance)
@@ -86,8 +77,7 @@ def parse(values: Mapping[str, object]) -> Settings:
SAVED_KEYS = ("SpeedLimitController", "ShowSpeedLimits", "IsMetric", "SLCPriority1", "SLCPriority2",
"SLCOverride", "SLCConfirmation", "SLCAdvanceSeconds",
*(f"Offset{i}" for i in range(1, 8)))
"SLCOverride", "SLCConfirmation", "SLCAdvanceSeconds")
def read_params(params) -> Settings:
@@ -105,7 +95,7 @@ def read_params(params) -> Settings:
except (ValueError, OverflowError):
return _disabled('Invalid saved SLCOffsetSchedule')
for key in SAVED_KEYS:
if raw_document is not None and (key == 'IsMetric' or key.startswith('Offset')):
if raw_document is not None and key == 'IsMetric':
continue
# Typed Params reads turn malformed numbers/strings into absence and accept
# any non-"1" Boolean as false. Neither may silently alter confirmation or
@@ -113,9 +103,7 @@ def read_params(params) -> Settings:
try:
with Path(params.get_param_path(key)).open('rb') as stream:
raw = stream.read(8 if key == 'SLCAdvanceSeconds' else -1)
if key.startswith('Offset'):
value = float(raw)
elif key == 'SLCAdvanceSeconds':
if key == 'SLCAdvanceSeconds':
if raw not in tuple(str(seconds).encode() for seconds in range(11)):
raise ValueError('invalid advance time')
value = int(raw)
@@ -1,4 +1,4 @@
"""The adopted SI schedule, legacy zero tail and invalid-source boundaries."""
"""Current SI schedules, zero defaults and invalid-source boundaries."""
from pathlib import Path
import tempfile
@@ -16,13 +16,13 @@ def offset_at(schedule, speed):
class OffsetDocumentTests(unittest.TestCase):
def test_all_legacy_edges_and_frozen_zero_tail(self):
def test_all_schedule_edges_and_zero_tail(self):
epsilon = 0.00001
for metric, bounds, conversion in ((False, od.IMPERIAL_BOUNDS, od.MPH_TO_MPS),
(True, od.METRIC_BOUNDS, od.KPH_TO_MPS)):
with self.subTest(metric=metric):
values = {'SpeedLimitController': True, 'IsMetric': metric,
**{f'Offset{i}': float(i) for i in range(1, 8)}}
document = od.OffsetDocument(bounds, tuple(index * conversion for index in range(1, 8)))
values = {'SpeedLimitController': True, 'SLCOffsetSchedule': document}
schedule = parse(values).offsets
self.assertEqual(len(schedule.bands), 8)
for index in range(1, 8):
@@ -36,7 +36,7 @@ class OffsetDocumentTests(unittest.TestCase):
self.assertEqual(schedule.bands[-1].offset_mps, 0.0)
def test_adopted_document_is_sole_numeric_authority(self):
document = od.adopt_legacy(False, (0, 1, -2, 3, 4, 5, 7))
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, tuple(value * od.MPH_TO_MPS for value in (0, 1, -2, 3, 4, 5, 7)))
encoded = od.encode(document)
self.assertEqual(od.decode(encoded), document)
expected = document.schedule()
@@ -79,7 +79,7 @@ class OffsetDocumentTests(unittest.TestCase):
params.put_bool('SpeedLimitController', True, block=True)
legacy = Path(params.get_param_path('Offset7'))
legacy.write_bytes(b'5.0000')
document = od.adopt_legacy(False, (0, 1, 2, 3, 4, 5, 5))
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, tuple(value * od.MPH_TO_MPS for value in (0, 1, 2, 3, 4, 5, 5)))
params.put('SLCOffsetSchedule', od.to_value(document), block=True)
self.assertEqual(params.get('SLCOffsetSchedule'), od.to_value(document))
self.assertEqual(read_params(params).offsets, document.schedule())
@@ -106,7 +106,7 @@ class OffsetDocumentTests(unittest.TestCase):
params.put_bool('SpeedLimitController', True, block=True)
Path(params.get_param_path('SLCOffsetSchedule')).write_bytes(raw)
self.assertFalse(read_params(params).enabled)
valid = od.adopt_legacy(True, (0, 0, 0, 0, 0, 0, 0))
valid = od.default_document(True)
for altered in ({'version': True, 'bounds_mps': valid.bounds_mps, 'offsets_mps': valid.offsets_mps},
{'version': 1, 'bounds_mps': (0, 1, 1, 3, 4, 5, 6, 7), 'offsets_mps': valid.offsets_mps},
{'version': 1, 'bounds_mps': valid.bounds_mps, 'offsets_mps': (0, 0, 0, 0, 0, 0, True)}):
@@ -1116,7 +1116,6 @@ class RuntimeReplayTests(unittest.TestCase):
params.put('SLCFallback', 2, block=True)
params.put('SLCPriority1', 'Map Data', block=True)
params.put('SLCPriority2', 'Dashboard', block=True)
params.put('Offset3', 5.0, block=True)
restored = read_params(params)
self.assertTrue(restored.enabled)
self.assertTrue(restored.acceptance.confirm_lower)
@@ -4,8 +4,9 @@ from pathlib import Path
import tempfile
import unittest
from openpilot.common.params import Params
from openpilot.common.params import Params, UnknownKeyName
from openpilot.starpilot.speed_limits import acceptance as acc
from openpilot.starpilot.speed_limits import offset_document as od
from openpilot.starpilot.speed_limits.runtime_settings import MPH_TO_MPS, parse, read_params
@@ -60,8 +61,7 @@ class RuntimeSettingsTests(unittest.TestCase):
self.assertEqual(Path(params.get_param_path('SLCPriority2')).read_bytes(), b'Map Data')
def test_corrupt_saved_values_disable_control_without_rewriting(self):
cases = {'SLCConfirmation': b'invalid',
'Offset3': b'bad', 'IsMetric': b'2', 'SLCPriority1': b'\xff'}
cases = {'SLCConfirmation': b'invalid', 'IsMetric': b'2', 'SLCPriority1': b'\xff'}
for key, raw in cases.items():
with self.subTest(key=key), tempfile.TemporaryDirectory() as directory:
params = Params(directory)
@@ -90,7 +90,8 @@ class RuntimeSettingsTests(unittest.TestCase):
params.put_bool('SLCConfirmation', True, block=True)
params.put_bool('SLCConfirmationLower', True, block=True)
params.put('SLCFallback', 0, block=True)
params.put('Offset3', -5.0, block=True)
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (0, 0, -5 * MPH_TO_MPS, 0, 0, 0, 0))
params.put('SLCOffsetSchedule', od.to_value(document), block=True)
settings = read_params(params)
self.assertTrue(settings.enabled)
self.assertFalse(settings.errors)
@@ -114,13 +115,32 @@ class RuntimeSettingsTests(unittest.TestCase):
for old in (0, 1, 2, None, 'broken'):
self.assertEqual(parse({'SpeedLimitController': True, 'SLCFallback': old}), baseline)
def test_nonfinite_or_unrepresentable_offset_disables_control(self):
def test_nonfinite_or_unrepresentable_current_offset_disables_control(self):
for value in (float('nan'), float('inf'), 10 ** 1000):
with self.subTest(kind=type(value).__name__):
settings = parse({'SpeedLimitController': True, 'Offset3': value})
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (0, 0, value, 0, 0, 0, 0))
settings = parse({'SpeedLimitController': True, 'SLCOffsetSchedule': document})
self.assertFalse(settings.enabled)
self.assertTrue(settings.errors)
def test_retired_offsets_are_unregistered_and_ignored_with_zero_defaults(self):
for metric in (False, True):
with self.subTest(metric=metric), tempfile.TemporaryDirectory() as directory:
params = Params(directory)
params.put_bool('SpeedLimitController', True, block=True)
params.put_bool('IsMetric', metric, block=True)
for index in range(1, 8):
key = f'Offset{index}'
with self.assertRaises(UnknownKeyName):
params.check_key(key)
Path(params.get_param_path(key)).write_bytes(b'3' if index % 2 else b'broken')
settings = read_params(params)
self.assertTrue(settings.enabled)
self.assertEqual(settings.offsets, od.default_document(metric).schedule())
self.assertFalse(Path(params.get_param_path('SLCOffsetSchedule')).exists())
for retired in (3, float('nan'), 'broken', 10 ** 1000):
self.assertEqual(parse({'SpeedLimitController': True, 'IsMetric': metric, 'Offset1': retired}), settings)
if __name__ == '__main__':
unittest.main()
@@ -418,15 +418,18 @@ class TestStateMigration(unittest.TestCase):
self.assertTrue(self.params.get_bool('StarPortEnabled'))
self.assertFalse(Path(self.params.get_param_path('AndroidAutoEnabled')).exists())
self.params.put_bool('AlwaysOnLateral', True, block=True)
corrupt = {'SLCFallback': b'broken', 'Offset3': b'',
corrupt = {'SLCFallback': b'broken', 'SLCOffsetSchedule': b'',
'LongitudinalPersonalityProfiles': b'{broken', 'LaneCenterOffset': b'bad'}
for key, raw in corrupt.items():
Path(self.params.get_param_path(key)).write_bytes(raw)
retired_offset = Path(self.params.get_param_path('Offset3'))
retired_offset.write_bytes(b'3')
manager.manager_init()
self.assertTrue(self.params.get_bool('AlwaysOnLateral'))
self.assertFalse(self.params.get_bool('OpenpilotEnabledToggle'))
for key, raw in corrupt.items():
self.assertEqual(Path(self.params.get_param_path(key)).read_bytes(), raw)
self.assertFalse(retired_offset.exists())
if __name__ == '__main__':
+1 -1
View File
@@ -210,7 +210,7 @@ class FeatureSettingsView:
elif row.page and value is None:
self._center("Open >", DETAIL_SIZE, action, TEXT_PRIMARY if row.available or row.choices == ("Off", "On") else TEXT_MUTED)
elif _confirmation(row):
if row.key in ("torque_adopt", "slc_adopt"):
if row.key == "torque_adopt":
action = "Adopt…"
elif row.key in ("torque_rebase", "torque_gain_rebase"):
action = "Review…"
@@ -117,7 +117,6 @@ class FeatureSettingsCompact:
"torque_gain_rebase": "Apply the saved steering response to the selected controller and current vehicle tune? " +
"Friction and other models stay unchanged.",
"torque_rebase": f"reapply saved torque values ({row.value}) with the current vehicle tune?",
"slc_adopt": "keep these offsets and speed ranges when units change? old saved values remain, but no longer control the offsets",
"slc_reset": f"reset speed-limit offsets to zero? {row.value}. " +
"saved control may resume if its switch is on; its switch will not change",
"curve_reset": "Erase learned curve data? Curve Speed Controller keeps its current On or Off choice.",
@@ -34,7 +34,7 @@ from openpilot.starpilot.longitudinal.profile_document import (
synchronise_profile_document_enabled,
)
from openpilot.starpilot.ui.feature_settings_state import (
FeaturePage, FeatureRow, FeatureSettingsRequest, FeatureSettingsState, SLC_CONFIRM_ACTIONS, TORQUE_CONFIRM_ACTIONS,
FeaturePage, FeatureRow, FeatureSettingsRequest, FeatureSettingsState, SLC_CONFIRM_ACTIONS, SLC_OFFSET_KEYS, TORQUE_CONFIRM_ACTIONS,
)
from openpilot.starpilot.speed_limits.runtime_settings import MPH_TO_MPS
from openpilot.starpilot.aol.vehicle import policy_for as aol_policy_for
@@ -644,7 +644,7 @@ class FeatureSettingsOwner:
return "Vision"
if key == "IsMetric":
return "0"
if key.startswith("Offset") or key in ("ForceStopDistanceOffset", "SLCAdvanceSeconds"):
if key in SLC_OFFSET_KEYS or key in ("ForceStopDistanceOffset", "SLCAdvanceSeconds"):
return "0"
value = self.params.get_default_value(key)
return str(value) if value is not None else ""
@@ -1443,7 +1443,7 @@ class FeatureSettingsOwner:
return self.traffic_profiles.apply(request)
if key in TORQUE_CONFIRM_ACTIONS or key.startswith("torque:"):
return self.torque.apply(request)
if key in SLC_CONFIRM_ACTIONS or key in ("Offset1", "Offset2", "Offset3", "Offset4", "Offset5", "Offset6", "Offset7"):
if key in SLC_CONFIRM_ACTIONS or key in SLC_OFFSET_KEYS:
return self.slc_offsets.apply(request)
if is_long_confirm_action(key):
return self.long_profiles.apply(request)
@@ -65,7 +65,8 @@ class FeaturePage(StrEnum):
TORQUE_CONFIRM_ACTIONS = frozenset(("torque_adopt", "torque_reset", "torque_reset_profile", "torque_rebase", "torque_prepare_firestar", "torque_gain_rebase"))
SLC_CONFIRM_ACTIONS = frozenset(("slc_adopt", "slc_reset"))
SLC_CONFIRM_ACTIONS = frozenset(("slc_reset",))
SLC_OFFSET_KEYS = tuple(f"slc:offset:{index}" for index in range(1, 8))
LANE_CHANGE_CONFIRM_ACTIONS = frozenset(("lane_change:reset",))
CURVE_CONFIRM_ACTIONS = frozenset(("curve_reset",))
CONDITIONAL_CONFIRM_ACTIONS = frozenset(("conditional:reset", "conditional:manual_reset"))
@@ -527,7 +528,7 @@ def row_change(row: FeatureRow, direction: int = 1) -> FeatureSettingsRequest |
"""Bind a press to the displayed source, never a later refreshed value."""
if not row.available:
return None
if row.key in ("Offset1", "Offset2", "Offset3", "Offset4", "Offset5", "Offset6", "Offset7"):
if row.key in SLC_OFFSET_KEYS:
if direction not in (-1, 1):
return None
return FeatureSettingsRequest(row.key, row.source, "", related_source=row.related_source,
+2 -3
View File
@@ -1184,7 +1184,7 @@ class StarShellSession:
from openpilot.starpilot.ui.feature_settings_state import CURVE_CONFIRM_ACTIONS, CONDITIONAL_CONFIRM_ACTIONS, TORQUE_CONFIRM_ACTIONS, is_long_confirm_action
from openpilot.starpilot.ui.long_profile_feature import long_confirm_question
from openpilot.starpilot.ui.controller_feature import SETUP_ACTION, SETUP_QUESTION
action_key = row.key if (row.key in TORQUE_CONFIRM_ACTIONS or row.key in ("slc_adopt", "slc_reset", LANE_CHANGE_RESET, SETUP_ACTION) or
action_key = row.key if (row.key in TORQUE_CONFIRM_ACTIONS or row.key in ("slc_reset", LANE_CHANGE_RESET, SETUP_ACTION) or
row.key in CURVE_CONFIRM_ACTIONS | CONDITIONAL_CONFIRM_ACTIONS or is_long_confirm_action(row.key)) else "reset_profiles"
if action_key in CONDITIONAL_CONFIRM_ACTIONS:
self._confirm_conditional(FeatureSettingsRequest(action_key, row.source, "confirm", confirmation=True,
@@ -1203,7 +1203,6 @@ class StarShellSession:
"Friction and other models stay unchanged.",
"torque_rebase": f"Reapply saved torque values ({row.value}) with the current vehicle tune? " +
"Saved values must pass current limits.",
"slc_adopt": "Keep these speed-limit offsets and speed ranges when units change? Legacy values remain saved but inactive.",
"slc_reset": f"Reset speed-limit offsets to zero? {row.value}. " +
"Saved SLC control may resume if its switch is On; the switch will not change.",
"curve_reset": "Reset saved Curve learning? Your Curve Speed Controller On/Off choice will stay the same.",
@@ -1214,7 +1213,7 @@ class StarShellSession:
related_source=row.related_source,
vehicle_fingerprint=row.vehicle_fingerprint,
capability=row.capability, dependencies=row.dependencies))
if action_key in ("torque_adopt", "slc_adopt"):
if action_key == "torque_adopt":
button = "Continue"
elif action_key in ("torque_rebase", "torque_gain_rebase"):
button = "Review"
+26 -116
View File
@@ -7,12 +7,11 @@ import math
import threading
from openpilot.starpilot.speed_limits import offset_document as od
from openpilot.starpilot.ui.feature_settings_state import FeatureRow, FeatureSettingsRequest
from openpilot.starpilot.ui.feature_settings_state import FeatureRow, FeatureSettingsRequest, SLC_OFFSET_KEYS
from openpilot.starpilot.saved_source import read_saved
from openpilot.starpilot.saved_document import commit_exact
DOCUMENT_KEY = "SLCOffsetSchedule"
OFFSET_KEYS = tuple(f"Offset{i}" for i in range(1, 8))
_LOCK = threading.RLock()
@@ -30,7 +29,6 @@ class OffsetSnapshot:
raw_document: bytes | None
raw_unit: bytes | None
raw_control: bytes | None
raw_legacy: tuple[bytes | None, ...]
document: od.OffsetDocument | od.NeedsReview | None
metric: bool | None
valid: bool
@@ -44,16 +42,6 @@ def _strict_unit(raw: bytes | None) -> bool | None:
return None
def _legacy(raw: tuple[bytes | None, ...], metric: bool) -> od.OffsetDocument:
values = []
for value in raw:
number = 0.0 if value is None else float(value)
if not math.isfinite(number):
raise ValueError("Invalid saved SLC offset")
values.append(number)
return od.adopt_legacy(metric, tuple(values))
def _unit_factor(metric: bool) -> float:
return od.KPH_TO_MPS if metric else od.MPH_TO_MPS
@@ -65,7 +53,7 @@ def _minimum_offset(document: od.OffsetDocument, index: int, metric: bool) -> fl
def _display(value: float) -> str:
return f"{value:.2f}".rstrip("0").rstrip(".")
return str(round(value))
def _capability(cp, configuration_longitudinal=False) -> tuple | None:
@@ -100,21 +88,20 @@ class SlcOffsetOwner:
raw_doc = self._raw(DOCUMENT_KEY)
raw_unit = self._raw("IsMetric")
raw_control = self._raw("SpeedLimitController")
raw_legacy = tuple(self._raw(key) for key in OFFSET_KEYS) if raw_doc is None else (None,) * 7
metric = _strict_unit(raw_unit)
try:
if not all(self._readable(key) for key in (DOCUMENT_KEY, "IsMetric", "SpeedLimitController")):
raise ValueError("Unreadable SLC setting")
if raw_doc is not None:
document = od.decode(raw_doc)
elif metric is not None and all(self._readable(key) for key in OFFSET_KEYS):
document = _legacy(raw_legacy, metric)
elif metric is not None:
document = od.default_document(metric)
else:
document = None
return OffsetSnapshot(raw_doc, raw_unit, raw_control, raw_legacy, document, metric,
return OffsetSnapshot(raw_doc, raw_unit, raw_control, document, metric,
document is not None and metric is not None)
except (ValueError, OSError, OverflowError):
return OffsetSnapshot(raw_doc, raw_unit, raw_control, raw_legacy, None, metric, False)
return OffsetSnapshot(raw_doc, raw_unit, raw_control, None, metric, False)
def ready_to_enable(self) -> bool:
snap = self.snapshot()
@@ -126,25 +113,13 @@ class SlcOffsetOwner:
available = allowed and capable is not None
repair_available = available if repair_allowed is None else repair_allowed and capable is not None
deps = (("IsMetric", snap.raw_unit), ("SpeedLimitController", snap.raw_control))
if snap.raw_document is None:
deps += tuple(zip(OFFSET_KEYS, snap.raw_legacy, strict=True))
if snap.raw_document is None and isinstance(snap.document, od.OffsetDocument):
mode = "Saved legacy schedule"
action = FeatureRow("slc_adopt", "Adopt fixed offsets", "Keep these offsets and speed ranges when units change", None,
available=repair_available, related_source=snap.raw_unit,
capability=capable, dependencies=deps)
elif isinstance(snap.document, od.OffsetDocument):
if isinstance(snap.document, od.OffsetDocument):
mode = "Fixed speed ranges"
action = None
elif snap.document is od.NEEDS_REVIEW:
mode = "Saved offsets need review; SLC control is paused"
action = None
else:
mode = "Invalid saved offsets; SLC control is paused"
action = None
rows = [FeatureRow("", "Offset policy", mode)]
if action is not None:
rows.append(action)
if isinstance(snap.document, od.OffsetDocument) and snap.metric is not None:
unit = "km/h" if snap.metric else "mph"
scale = 1 / _unit_factor(snap.metric)
@@ -153,17 +128,16 @@ class SlcOffsetOwner:
# Round presentation only: the stored SI boundaries remain authoritative.
label = f"{display_bounds[index]}–<{display_bounds[index + 1]} {unit}"
value = _display(snap.document.offsets_mps[index] * scale)
rows.append(FeatureRow(OFFSET_KEYS[index], label + " offset", value, snap.raw_document,
step=1 if snap.raw_document is not None else 0, unit=unit,
rows.append(FeatureRow(SLC_OFFSET_KEYS[index], label + " offset", value, snap.raw_document,
step=1, unit=unit,
minimum=_minimum_offset(snap.document, index, snap.metric), maximum=150 if snap.metric else 99,
available=available and snap.raw_document is not None,
reason="Legacy value; adopt to edit" if snap.raw_document is None else
"Upper boundary excluded; displayed ranges rounded to whole speeds. Unsafe reductions use the unadjusted limit",
available=available,
reason="Upper boundary excluded; displayed ranges rounded to whole speeds. Unsafe reductions use the unadjusted limit",
related_source=snap.raw_unit, capability=capable, dependencies=deps,
display_unit=unit, default_value="0"))
rows.append(FeatureRow("", f"{display_bounds[-1]} {unit} and above", "Zero offset"))
else:
for index, key in enumerate(OFFSET_KEYS, start=1):
for index, key in enumerate(SLC_OFFSET_KEYS, start=1):
label = f"Band {index} offset (invalid saved units)" if snap.metric is None else f"Band {index} offset"
rows.append(FeatureRow(key, label, "Invalid saved offset or document", snap.raw_document,
reason="Unavailable until saved source is repaired"))
@@ -181,9 +155,7 @@ class SlcOffsetOwner:
def _same(self, request: FeatureSettingsRequest, snap: OffsetSnapshot, *, require_capability: bool) -> bool:
expected_deps = (("IsMetric", snap.raw_unit), ("SpeedLimitController", snap.raw_control))
if snap.raw_document is None:
expected_deps += tuple(zip(OFFSET_KEYS, snap.raw_legacy, strict=True))
authority = self.repair_parked if request.key in ("slc_adopt", "slc_reset") else self.parked
authority = self.repair_parked if request.key == "slc_reset" else self.parked
return (authority() and (not require_capability or request.capability is not None and request.capability == self.capability()) and
request.dependencies == expected_deps and
self._raw(DOCUMENT_KEY) == request.expected and self._readable(DOCUMENT_KEY) and
@@ -192,37 +164,31 @@ class SlcOffsetOwner:
snap.raw_control == self._raw("SpeedLimitController"))
def apply(self, request: FeatureSettingsRequest) -> bool:
if request.key not in ("slc_adopt", "slc_reset", *OFFSET_KEYS):
if request.key not in ("slc_reset", *SLC_OFFSET_KEYS):
return False
with _LOCK:
snap = self.snapshot()
if not self._same(request, snap, require_capability=True):
return False
try:
if request.key == "slc_adopt":
if not request.confirmation or request.expected is not None or not isinstance(snap.document, od.OffsetDocument):
return False
result = snap.document
elif request.key == "slc_reset":
if request.key == "slc_reset":
if not request.confirmation or snap.metric is None or \
request.expected is None and isinstance(snap.document, od.OffsetDocument):
return False
if isinstance(snap.document, od.OffsetDocument):
result = od.OffsetDocument(snap.document.bounds_mps, (0.0,) * 7)
else:
result = od.adopt_legacy(snap.metric, (0.0,) * 7)
result = od.default_document(snap.metric)
else:
if not isinstance(snap.document, od.OffsetDocument) or request.expected is None or snap.metric is None or \
if not isinstance(snap.document, od.OffsetDocument) or snap.metric is None or \
request.direction not in (-1, 0, 1) or request.display_unit != ("km/h" if snap.metric else "mph"):
return False
if request.key not in OFFSET_KEYS:
return False
index = OFFSET_KEYS.index(request.key)
index = SLC_OFFSET_KEYS.index(request.key)
current = snap.document.offsets_mps[index]
low, high = _minimum_offset(snap.document, index, snap.metric), 150.0 if snap.metric else 99.0
factor = _unit_factor(snap.metric)
candidate = (float(request.value) * factor if request.direction == 0 else
max(low * factor, min(high * factor, current + request.direction * factor)))
candidate = (round(float(request.value)) * factor if request.direction == 0 else
max(low, min(high, round(current / factor) + request.direction)) * factor)
if not math.isfinite(candidate) or not low * factor <= candidate <= high * factor:
return False
new_offsets = list(snap.document.offsets_mps)
@@ -238,68 +204,12 @@ class SlcOffsetOwner:
return False
def change_units(self, desired: bool, expected_unit: bytes | None) -> bool:
"""Preserve an adopted SI schedule before the global presentation change."""
"""Change display units without changing the saved physical offsets."""
if type(desired) is not bool:
return False
with _LOCK:
snap = self.snapshot()
if not self._readable("IsMetric") or snap.raw_unit != expected_unit:
return False
if snap.metric is not None and snap.metric == desired:
return True
if (snap.raw_document is None and snap.raw_control in (None, b"0") and
isinstance(snap.document, od.OffsetDocument) and not any(snap.document.offsets_mps)):
result = commit_exact(self.params, key="IsMetric", max_bytes=8,
raw=b"1" if desired else b"0", expected=expected_unit,
authorized=lambda: self._same_unit_source(snap, None), temp_prefix=".display-units-")
return result.committed and result.verified
try:
protected_capability = None
migrating = False
written_document = None
if snap.raw_document is None:
if isinstance(snap.document, od.OffsetDocument):
if not self.repair_parked():
return False
protected_capability = self.capability()
migrating = True
if not self._same_unit_source(snap, protected_capability):
return False
self.params.put(DOCUMENT_KEY, od.to_value(snap.document), block=True)
written_document = snap.document
elif snap.raw_control not in (None, b"0"):
if not self.repair_parked() or not self._same_unit_source(snap, self.capability()):
return False
protected_capability = self.capability()
migrating = True
self.params.put(DOCUMENT_KEY, od.to_value(od.NEEDS_REVIEW), block=True)
written_document = od.NEEDS_REVIEW
if not self._readable("IsMetric") or self._raw("IsMetric") != expected_unit:
return False
if snap.raw_document is None:
written = self._raw(DOCUMENT_KEY)
if written_document is not None:
if written is None or not self._readable(DOCUMENT_KEY) or od.decode(written) != written_document:
return False
elif written is not None or not self._readable(DOCUMENT_KEY):
return False
elif self._raw(DOCUMENT_KEY) != snap.raw_document or not self._readable(DOCUMENT_KEY):
return False
if (snap.raw_document is None and
(not self._readable("SpeedLimitController") or self._raw("SpeedLimitController") != snap.raw_control) or
migrating and not self.repair_parked() or
protected_capability is not None and self.capability() != protected_capability):
return False
if snap.raw_document is None and written_document is None and not self._same_unit_source(snap, None):
return False
self.params.put_bool("IsMetric", desired, block=True)
return self._raw("IsMetric") == (b"1" if desired else b"0")
except (OSError, ValueError, TypeError, OverflowError):
return False
def _same_unit_source(self, snap: OffsetSnapshot, capability: tuple | None) -> bool:
return (self._readable(DOCUMENT_KEY) and self._raw(DOCUMENT_KEY) == snap.raw_document and
self._readable("IsMetric") and self._raw("IsMetric") == snap.raw_unit and
self._readable("SpeedLimitController") and self._raw("SpeedLimitController") == snap.raw_control and
all(self._readable(key) and self._raw(key) == raw for key, raw in zip(OFFSET_KEYS, snap.raw_legacy, strict=True)) and
(capability is None or self.repair_parked() and self.capability() == capability))
result = commit_exact(self.params, key="IsMetric", max_bytes=8,
raw=b"1" if desired else b"0", expected=expected_unit,
authorized=lambda: self._readable("IsMetric") and self._raw("IsMetric") == expected_unit,
temp_prefix=".display-units-")
return result.committed and result.verified
@@ -522,11 +522,11 @@ class FeatureSettingsOwnerTests(unittest.TestCase):
self.assertEqual(authority.value, "100.0")
self.assertTrue(self.owner.apply(required_change(authority, -1)))
self.assertAlmostEqual(self.params.get("LaneCenteringE2EAuthority"), 0.95)
offset = self._row("slc", "Offset1")
self.assertFalse(offset.available)
self.assertFalse(self.owner.apply(FeatureSettingsRequest("Offset1", offset.source, "5", vehicle_fingerprint=offset.vehicle_fingerprint)))
offset = self._row("slc", "slc:offset:1")
self.assertTrue(offset.available)
self.assertFalse(self.owner.apply(FeatureSettingsRequest("slc:offset:1", offset.source, "5", vehicle_fingerprint=offset.vehicle_fingerprint)))
Path(self.params.get_param_path("IsMetric")).write_bytes(b"bad")
offset = self._row("slc", "Offset1")
offset = self._row("slc", "slc:offset:1")
self.assertIn("invalid saved units", offset.label)
self.assertEqual(Path(self.params.get_param_path("IsMetric")).read_bytes(), b"bad")
@@ -51,14 +51,15 @@ class SlcOffsetFeatureTests(unittest.TestCase):
self.assertFalse(native_parked(ui))
factory.assert_called_once_with(ui)
def test_pcm_cruise_longitudinal_can_adopt_and_enable_slc(self):
def test_pcm_cruise_longitudinal_can_edit_and_enable_slc_without_adoption(self):
assert self.cp is not None
self.cp.carFingerprint = "TOYOTA_HIGHLANDER_TSS2"
self.cp.pcmCruise = True
state = self.owner.snapshot("slc", parked=True, system_long=False, lateral_context=True, metric=False)
adopt = next(row for row in state.rows if row.key == "slc_adopt")
self.assertTrue(adopt.available)
self.assertTrue(self.owner.apply(self.confirm(adopt)))
offset = self.row("slc:offset:1")
self.assertTrue(offset.available)
edit = row_change(offset)
assert edit is not None
self.assertTrue(self.owner.apply(edit))
row = self.row("SpeedLimitController")
self.assertTrue(row.available)
request = row_change(row)
@@ -67,33 +68,45 @@ class SlcOffsetFeatureTests(unittest.TestCase):
self.cp.openpilotLongitudinalControl = False
self.assertFalse(self.row("SpeedLimitController").available)
def test_valid_legacy_enable_preserves_sources_and_invalid_offsets_deny(self):
for legacy in (None, b"5"):
with self.subTest(legacy=legacy):
self.params.put_bool("SpeedLimitController", False, block=True)
if legacy is not None:
Path(self.params.get_param_path("Offset1")).write_bytes(legacy)
original = tuple(self.raw(key) for key in ("IsMetric", *(f"Offset{i}" for i in range(1, 8))))
row = self.row("SpeedLimitController")
self.assertTrue(row.available)
request = row_change(row)
assert request is not None
self.assertTrue(self.owner.apply(request))
self.assertTrue(read_params(self.params).enabled)
self.assertIsNone(self.raw(DOCUMENT_KEY))
self.assertEqual(tuple(self.raw(key) for key in ("IsMetric", *(f"Offset{i}" for i in range(1, 8)))), original)
for key, raw in (("Offset1", b"bad"), (DOCUMENT_KEY, b"{bad"),
(DOCUMENT_KEY, od.encode(od.NEEDS_REVIEW)), ("IsMetric", b"bad")):
def test_fresh_and_retired_offsets_start_at_editable_zero_without_adoption(self):
for metric in (False, True):
for retired in (None, b"5", b"bad"):
with self.subTest(metric=metric, retired=retired):
self.params.remove(DOCUMENT_KEY)
self.params.put_bool("IsMetric", metric, block=True)
Path(self.params.get_param_path("Offset1")).unlink(missing_ok=True)
if retired is not None:
Path(self.params.get_param_path("Offset1")).write_bytes(retired)
state = self.owner.snapshot("slc", parked=True, system_long=True, lateral_context=True, metric=metric)
offsets = [row for row in state.rows if row.key.startswith("slc:offset:")]
self.assertEqual(len(offsets), 7)
self.assertTrue(all(row.available and row.value == "0" for row in offsets))
self.assertFalse(any(row.key == "slc_adopt" for row in state.rows))
self.assertTrue(self.owner.slc_offsets.ready_to_enable())
self.assertTrue(all(band.offset_mps == 0 for band in read_params(self.params).offsets.bands))
self.assertFalse(self.owner.apply(FeatureSettingsRequest("slc_adopt", None, "confirm", confirmation=True)))
self.assertFalse(self.owner.apply(FeatureSettingsRequest("Offset1", None, "3")))
edit = row_change(offsets[0])
assert edit is not None
self.assertTrue(self.owner.apply(replace(edit, value="3", direction=0)))
document = od.decode(self.raw(DOCUMENT_KEY))
self.assertEqual(document.bounds_mps, od.default_document(metric).bounds_mps)
factor = od.KPH_TO_MPS if metric else od.MPH_TO_MPS
self.assertAlmostEqual(document.offsets_mps[0], 3 * factor)
self.assertEqual(read_params(self.params).offsets, document.schedule())
self.assertEqual(self.raw("Offset1"), retired)
def test_invalid_current_schedule_or_units_require_repair(self):
for key, raw in ((DOCUMENT_KEY, b"{bad"), (DOCUMENT_KEY, od.encode(od.NEEDS_REVIEW)), ("IsMetric", b"bad")):
with self.subTest(key=key, raw=raw):
for saved in ("Offset1", DOCUMENT_KEY, "IsMetric"):
path = Path(self.params.get_param_path(saved))
path.unlink(missing_ok=True)
self.params.put_bool("SpeedLimitController", False, block=True)
for saved in (DOCUMENT_KEY, "IsMetric"):
Path(self.params.get_param_path(saved)).unlink(missing_ok=True)
Path(self.params.get_param_path(key)).write_bytes(raw)
row = self.row("SpeedLimitController")
self.assertFalse(row.available)
self.assertIn("Repair invalid", row.reason)
self.assertFalse(self.owner.slc_offsets.ready_to_enable())
self.assertFalse(self.row("slc:offset:1").available)
def test_unavailable_control_explains_vehicle_and_context_gate(self):
assert self.cp is not None
@@ -107,49 +120,11 @@ class SlcOffsetFeatureTests(unittest.TestCase):
self.parked = False
self.assertIn("settings context is unavailable", self.row("SpeedLimitController").reason)
def test_explicit_adopt_edit_and_global_units_preserve_exact_si(self):
self.params.put_bool("IsMetric", False, block=True)
self.params.put("Offset1", 5.0, block=True)
self.params.put("Offset7", -2.0, block=True)
old = (self.raw("Offset1"), self.raw("Offset7"))
before = read_params(self.params).offsets
self.assertTrue(self.row("SpeedLimitController").available)
adopt = self.row("slc_adopt")
self.assertFalse(self.owner.apply(FeatureSettingsRequest(adopt.key, adopt.source, "confirm",
vehicle_fingerprint=adopt.vehicle_fingerprint,
capability=adopt.capability, dependencies=adopt.dependencies)))
self.assertTrue(self.owner.apply(self.confirm(adopt)))
self.assertTrue(self.row("SpeedLimitController").available)
self.assertEqual(read_params(self.params).offsets, before)
self.assertEqual((self.raw("Offset1"), self.raw("Offset7")), old)
first = self.row("Offset1")
self.assertTrue(first.available)
self.assertEqual(row_change(first).direction, 1)
first_edit = row_change(first)
assert first_edit is not None
self.assertTrue(self.owner.apply(first_edit))
document = od.decode(self.raw(DOCUMENT_KEY))
self.assertAlmostEqual(document.offsets_mps[0], 6 * od.MPH_TO_MPS)
self.assertEqual(document.schedule().bands[-1].offset_mps, 0.0)
self.assertTrue(self.units.change_units(True, b"0"))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), document)
metric_row = self.row("Offset1")
self.assertIn("km/h", metric_row.label)
metric_edit = row_change(metric_row)
assert metric_edit is not None
self.assertTrue(self.owner.apply(metric_edit))
adjusted = od.decode(self.raw(DOCUMENT_KEY))
self.assertAlmostEqual(adjusted.offsets_mps[0] - document.offsets_mps[0], od.KPH_TO_MPS)
self.assertTrue(self.units.change_units(False, b"1"))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), adjusted)
self.assertEqual((self.raw("Offset1"), self.raw("Offset7")), old)
def test_range_labels_follow_saved_si_bounds_without_rewriting_them(self):
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (0.0,) * 7)
self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
original = self.raw(DOCUMENT_KEY)
rows = [self.row(f"Offset{index}").label for index in range(1, 8)]
rows = [self.row(f"slc:offset:{index}").label for index in range(1, 8)]
self.assertEqual(rows, [f"{low}–<{high} mph offset" for low, high in
(("0", "25"), ("25", "34"), ("34", "44"), ("44", "54"),
("54", "64"), ("64", "74"), ("74", "99"))])
@@ -158,7 +133,7 @@ class SlcOffsetFeatureTests(unittest.TestCase):
self.assertEqual(self.raw(DOCUMENT_KEY), original)
self.assertTrue(self.units.change_units(True, None))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), document)
self.assertEqual(self.row("Offset4").label, "71–<87 km/h offset")
self.assertEqual(self.row("slc:offset:4").label, "71–<87 km/h offset")
def test_editor_rejects_offsets_that_cannot_produce_a_positive_target_in_the_band(self):
for metric, offset in ((False, -30), (True, -45)):
@@ -167,7 +142,7 @@ class SlcOffsetFeatureTests(unittest.TestCase):
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (0.0,) * 7)
self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
original = self.raw(DOCUMENT_KEY)
row = self.row("Offset1")
row = self.row("slc:offset:1")
request = row_change(row)
assert request is not None
self.assertGreater(row.minimum, offset)
@@ -179,7 +154,7 @@ class SlcOffsetFeatureTests(unittest.TestCase):
def test_stepper_can_repair_an_unsafe_saved_offset_and_clamps_at_the_band_minimum(self):
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (-30 * od.MPH_TO_MPS, 0, 0, 0, 0, 0, 0))
self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
row = self.row('Offset1')
row = self.row('slc:offset:1')
request = row_change(row)
assert request is not None
self.assertTrue(self.owner.apply(request))
@@ -187,7 +162,7 @@ class SlcOffsetFeatureTests(unittest.TestCase):
# Unit conversion can leave a fractional value less than one step above the minimum.
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, ((row.minimum + .5) * od.MPH_TO_MPS, 0, 0, 0, 0, 0, 0))
self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
request = row_change(self.row('Offset1'), -1)
request = row_change(self.row('slc:offset:1'), -1)
assert request is not None
self.assertTrue(self.owner.apply(request))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)).offsets_mps[0], row.minimum * od.MPH_TO_MPS)
@@ -201,7 +176,7 @@ class SlcOffsetFeatureTests(unittest.TestCase):
vehicle_params=lambda: self.cp)
state = owner.snapshot("slc", parked=False, system_long=True, lateral_context=True, metric=False,
configure_while_driving=True)
offset = next(row for row in state.rows if row.key == "Offset2")
offset = next(row for row in state.rows if row.key == "slc:offset:2")
reset = next(row for row in state.rows if row.key == "slc_reset")
self.assertTrue(offset.available)
self.assertFalse(reset.available)
@@ -213,21 +188,88 @@ class SlcOffsetFeatureTests(unittest.TestCase):
self.params.put_bool("IsMetric", True, block=True)
self.assertFalse(owner.apply(request))
def test_global_units_do_not_require_openpilot_longitudinal_or_a_fingerprint(self):
def test_global_units_do_not_require_longitudinal_a_fingerprint_or_migration(self):
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (5 * od.MPH_TO_MPS,) * 7)
for cp in (None, NS(carFingerprint="STOCK_CRUISE", openpilotLongitudinalControl=False,
notCar=False, dashcamOnly=False, passive=False)):
with self.subTest(vehicle=cp):
self.cp = cp
self.params.remove(DOCUMENT_KEY)
self.params.put_bool("IsMetric", False, block=True)
self.params.put("Offset1", 5.0, block=True)
self.assertTrue(self.units.change_units(True, b"0"))
document = od.decode(self.raw(DOCUMENT_KEY))
self.assertEqual(document, od.adopt_legacy(False, (5.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0)))
self.assertEqual(self.raw("IsMetric"), b"1")
self.assertEqual(self.raw("Offset1"), b"5.0")
for saved in (None, document):
with self.subTest(vehicle=cp, saved=saved):
self.cp = cp
self.parked = False
self.params.remove(DOCUMENT_KEY)
if saved is not None:
self.params.put(DOCUMENT_KEY, od.to_value(saved), block=True)
original = self.raw(DOCUMENT_KEY)
self.params.put_bool("IsMetric", False, block=True)
Path(self.params.get_param_path("Offset1")).write_bytes(b"5.0")
self.assertTrue(self.units.change_units(True, b"0"))
self.assertEqual(self.raw(DOCUMENT_KEY), original)
self.assertEqual(self.raw("IsMetric"), b"1")
self.assertEqual(self.raw("Offset1"), b"5.0")
def test_off_and_display_only_corrupt_legacy_allow_unit_change(self):
def test_first_edit_and_unit_changes_preserve_exact_physical_offsets(self):
edit = row_change(self.row("slc:offset:1"))
assert edit is not None
self.assertTrue(self.owner.apply(edit))
document = od.decode(self.raw(DOCUMENT_KEY))
self.assertEqual(document.offsets_mps[0], od.MPH_TO_MPS)
self.assertTrue(self.units.change_units(True, None))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), document)
metric_edit = row_change(self.row("slc:offset:1"))
assert metric_edit is not None
self.assertTrue(self.owner.apply(metric_edit))
adjusted = od.decode(self.raw(DOCUMENT_KEY))
self.assertAlmostEqual(adjusted.offsets_mps[0], 3 * od.KPH_TO_MPS)
self.assertTrue(self.units.change_units(False, b"1"))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), adjusted)
def test_first_edit_rejects_a_stale_missing_schedule(self):
edit = row_change(self.row("slc:offset:1"))
assert edit is not None
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (5 * od.MPH_TO_MPS,) * 7)
self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
self.assertFalse(self.owner.apply(edit))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), document)
def test_converted_offsets_use_whole_units_without_round_trip_drift(self):
cases = ((False, 3, 5), (True, 5, 3), (True, 4, 2), (False, -3, -5), (True, -5, -3))
for metric, original, converted in cases:
for direction in (-1, 1):
with self.subTest(metric=metric, original=original, direction=direction):
factor = od.KPH_TO_MPS if metric else od.MPH_TO_MPS
other_factor = od.MPH_TO_MPS if metric else od.KPH_TO_MPS
document = od.OffsetDocument(od.default_document(metric).bounds_mps, (original * factor,) * 7)
self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
self.params.put_bool("IsMetric", metric, block=True)
saved = self.raw(DOCUMENT_KEY)
for _ in range(3):
self.assertTrue(self.units.change_units(not metric, b"1" if metric else b"0"))
self.assertEqual(self.row("slc:offset:1").value, str(converted))
self.assertTrue(self.units.change_units(metric, b"0" if metric else b"1"))
self.assertEqual(self.row("slc:offset:1").value, str(original))
self.assertEqual(self.raw(DOCUMENT_KEY), saved)
self.assertTrue(self.units.change_units(not metric, b"1" if metric else b"0"))
edit = row_change(self.row("slc:offset:1"), direction)
assert edit is not None
self.assertTrue(self.owner.apply(edit))
changed = od.decode(self.raw(DOCUMENT_KEY))
self.assertAlmostEqual(changed.offsets_mps[0], (converted + direction) * other_factor)
self.assertEqual(changed.offsets_mps[1:], document.offsets_mps[1:])
edit = row_change(self.row("slc:offset:1"))
assert edit is not None
entry = replace(edit, direction=0, value="4.83")
self.assertTrue(self.owner.apply(entry))
self.assertAlmostEqual(od.decode(self.raw(DOCUMENT_KEY)).offsets_mps[0], 5 * other_factor)
def test_failed_unit_write_preserves_current_schedule_and_units(self):
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (5 * od.MPH_TO_MPS,) * 7)
self.params.put(DOCUMENT_KEY, od.to_value(document), block=True)
with patch("openpilot.starpilot.saved_document.os.replace", side_effect=OSError("write failed")):
self.assertFalse(self.units.change_units(True, None))
self.assertIsNone(self.raw("IsMetric"))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), document)
def test_retired_offset_corruption_does_not_block_unit_changes(self):
Path(self.params.get_param_path("Offset1")).write_bytes(b"nan")
Path(self.params.get_param_path("IsMetric")).write_bytes(b"bad")
self.params.put_bool("ShowSpeedLimits", True, block=True)
@@ -238,102 +280,8 @@ class SlcOffsetFeatureTests(unittest.TestCase):
self.assertEqual(self.raw("Offset1"), b"nan")
self.assertEqual(self.raw("IsMetric"), b"1")
def test_valid_legacy_unit_migration_waits_for_park_even_with_slc_off(self):
self.params.put_bool("IsMetric", False, block=True)
self.params.put("Offset1", 5.0, block=True)
self.params.put_bool("SpeedLimitController", False, block=True)
self.parked = False
self.assertFalse(self.units.change_units(True, b"0"))
self.assertIsNone(self.raw(DOCUMENT_KEY))
self.assertEqual(self.raw("IsMetric"), b"0")
def test_saved_on_corrupt_legacy_writes_marker_before_unit(self):
self.params.put_bool("SpeedLimitController", True, block=True)
Path(self.params.get_param_path("Offset3")).write_bytes(b"bad")
self.assertTrue(self.units.change_units(True, None))
self.assertIs(od.decode(self.raw(DOCUMENT_KEY)), od.NEEDS_REVIEW)
self.assertEqual(self.raw("IsMetric"), b"1")
self.assertFalse(read_params(self.params).enabled)
reset = self.row("slc_reset")
self.assertTrue(self.owner.apply(self.confirm(reset)))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)).offsets_mps, (0.0,) * 7)
self.assertTrue(read_params(self.params).enabled)
self.assertEqual(self.raw("Offset3"), b"bad")
def test_marker_failure_veto_and_stale_source_recheck(self):
self.params.put_bool("SpeedLimitController", True, block=True)
Path(self.params.get_param_path("Offset1")).write_bytes(b"bad")
original = self.params.put
def fails_marker(key, value, block=False):
if key == DOCUMENT_KEY:
raise OSError("write failed")
return original(key, value, block=block)
with patch.object(self.params, "put", side_effect=fails_marker):
self.assertFalse(self.units.change_units(True, None))
self.assertIsNone(self.raw("IsMetric"))
self.assertIsNone(self.raw(DOCUMENT_KEY))
Path(self.params.get_param_path("Offset1")).write_bytes(b"1")
adopt = self.row("slc_adopt")
Path(self.params.get_param_path("Offset1")).write_bytes(b"2")
self.assertFalse(self.owner.apply(self.confirm(adopt)))
self.assertIsNone(self.raw(DOCUMENT_KEY))
def test_absent_corrupt_legacy_reset_recovers_without_changing_saved_switch(self):
for saved_on in (False, True):
with self.subTest(saved_on=saved_on):
if saved_on:
self.params.put_bool("SpeedLimitController", True, block=True)
Path(self.params.get_param_path("Offset1")).write_bytes(b"bad")
self.assertFalse(read_params(self.params).enabled)
reset = self.row("slc_reset")
self.assertTrue(reset.available)
self.assertIsNone(reset.source)
self.assertFalse(self.owner.apply(FeatureSettingsRequest("slc_reset", None, "confirm",
confirmation=False, related_source=reset.related_source,
capability=reset.capability, dependencies=reset.dependencies)))
self.assertTrue(self.owner.apply(self.confirm(reset)))
document = od.decode(self.raw(DOCUMENT_KEY))
self.assertEqual(document, od.adopt_legacy(False, (0.0,) * 7))
self.assertEqual(self.raw("Offset1"), b"bad")
self.assertEqual(self.raw("SpeedLimitController"), b"1" if saved_on else None)
self.assertEqual(read_params(self.params).enabled, saved_on)
Path(self.params.get_param_path(DOCUMENT_KEY)).unlink()
def test_absent_corrupt_reset_rechecks_legacy_and_waits_for_valid_units(self):
legacy = Path(self.params.get_param_path("Offset4"))
legacy.write_bytes(b"bad")
reset = self.row("slc_reset")
legacy.write_bytes(b"different bad")
self.assertFalse(self.owner.apply(self.confirm(reset)))
self.assertIsNone(self.raw(DOCUMENT_KEY))
Path(self.params.get_param_path("IsMetric")).write_bytes(b"invalid")
self.assertFalse(self.row("slc_reset").available)
self.assertFalse(self.owner.apply(self.confirm(self.row("slc_reset"))))
self.assertIsNone(self.raw(DOCUMENT_KEY))
self.assertTrue(self.units.change_units(True, b"invalid"))
self.assertTrue(self.row("slc_reset").available)
self.assertTrue(self.owner.apply(self.confirm(self.row("slc_reset"))))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)).bounds_mps, od.METRIC_BOUNDS)
def test_saved_off_corrupt_unit_change_rechecks_legacy_before_unit_write(self):
legacy = Path(self.params.get_param_path("Offset1"))
legacy.write_bytes(b"bad")
original = self.units._same_unit_source
def mutate_at_final_guard(snap, capability):
legacy.write_bytes(b"changed")
return original(snap, capability)
with patch.object(self.units, "_same_unit_source", side_effect=mutate_at_final_guard):
self.assertFalse(self.units.change_units(True, None))
self.assertIsNone(self.raw("IsMetric"))
self.assertIsNone(self.raw(DOCUMENT_KEY))
def test_stale_unit_capability_and_parked_edit_write_nothing(self):
self.assertTrue(self.owner.apply(self.confirm(self.row("slc_adopt"))))
edit = row_change(self.row("Offset2"))
edit = row_change(self.row("slc:offset:2"))
assert edit is not None
self.params.put_bool("IsMetric", True, block=True)
self.assertFalse(self.owner.apply(edit))
@@ -346,7 +294,7 @@ class SlcOffsetFeatureTests(unittest.TestCase):
self.assertFalse(self.owner.apply(edit))
def test_valid_document_ignores_legacy_bytes_and_reset_keeps_bounds(self):
self.assertTrue(self.owner.apply(self.confirm(self.row("slc_adopt"))))
self.params.put(DOCUMENT_KEY, od.to_value(od.default_document(False)), block=True)
original = od.decode(self.raw(DOCUMENT_KEY))
Path(self.params.get_param_path("Offset1")).write_bytes(b"bad")
self.assertTrue(self.units.change_units(True, None))
@@ -355,16 +303,6 @@ class SlcOffsetFeatureTests(unittest.TestCase):
self.assertTrue(self.owner.apply(self.confirm(reset)))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)).bounds_mps, original.bounds_mps)
def test_unit_write_failure_after_adoption_keeps_original_unit_and_si_schedule(self):
self.params.put_bool("SpeedLimitController", True, block=True)
self.params.put("Offset2", 3.0, block=True)
expected = od.adopt_legacy(False, (0.0, 3.0, 0.0, 0.0, 0.0, 0.0, 0.0))
with patch.object(self.params, "put_bool", side_effect=OSError("unit write failed")):
self.assertFalse(self.units.change_units(True, None))
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), expected)
self.assertIsNone(self.raw("IsMetric"))
self.assertEqual(self.raw("Offset2"), b"3.0")
def test_corrupt_document_and_units_require_explicit_repairs(self):
self.params.put_bool("SpeedLimitController", True, block=True)
Path(self.params.get_param_path(DOCUMENT_KEY)).write_bytes(b"{broken")
@@ -378,9 +316,8 @@ class SlcOffsetFeatureTests(unittest.TestCase):
self.assertTrue(read_params(self.params).enabled)
def test_all_seven_edits_and_stale_unit_request(self):
self.assertTrue(self.owner.apply(self.confirm(self.row("slc_adopt"))))
for index in range(1, 8):
row = self.row(f"Offset{index}")
row = self.row(f"slc:offset:{index}")
edit = row_change(row, -1)
assert edit is not None
self.assertTrue(self.owner.apply(edit))
@@ -396,7 +333,7 @@ class SlcOffsetFeatureTests(unittest.TestCase):
from openpilot.selfdrive.ui.mici.layouts.settings import toggles as compact
self.cp = None
self.params.put("Offset1", 5.0, block=True)
self.params.put(DOCUMENT_KEY, od.to_value(od.default_document(False)), block=True)
large = TogglesLayout.__new__(TogglesLayout)
large._params = self.params
large._slc_offsets = self.units
@@ -424,7 +361,7 @@ class SlcOffsetFeatureTests(unittest.TestCase):
self.assertEqual(self.raw("IsMetric"), b"0")
self.assertEqual(od.decode(self.raw(DOCUMENT_KEY)), adopted)
def test_both_native_adapters_confirm_adoption_with_displayed_unit_source(self):
def test_both_native_reset_confirmations_recover_current_document_and_recheck_source(self):
from openpilot.starpilot.ui import feature_settings_compact as compact
from openpilot.starpilot.ui.runtime_app import StarShellSession
from openpilot.system.ui.widgets import DialogResult
@@ -433,48 +370,17 @@ class SlcOffsetFeatureTests(unittest.TestCase):
def __init__(self, _question, *_args, callback=None, **_kwargs):
self.confirmed = callback or _args[-1]
self.params.put_bool("IsMetric", True, block=True)
row = self.row("slc_adopt")
large = StarShellSession.__new__(StarShellSession)
cast(Any, large).feature_request = self.owner.apply
pushed = []
with patch("openpilot.system.ui.widgets.confirm_dialog.ConfirmDialog", Dialog), \
patch("openpilot.starpilot.ui.runtime_app.gui_app.push_widget", pushed.append):
large._confirm_feature_reset(row)
pushed[-1].confirmed(DialogResult.CONFIRM)
self.assertIsInstance(od.decode(self.raw(DOCUMENT_KEY)), od.OffsetDocument)
Path(self.params.get_param_path(DOCUMENT_KEY)).unlink()
row = self.row("slc_adopt")
compact_adapter = compact.FeatureSettingsCompact(cast(Any, NS(feature_request=self.owner.apply)))
pushed = []
with patch.object(compact, "BigConfirmationDialog", Dialog), \
patch.object(compact.gui_app, "push_widget", pushed.append), \
patch.object(compact.gui_app, "texture", lambda *_args: None):
compact_adapter._confirm_reset(row, lambda: None)
pushed[-1].confirmed()
self.assertIsInstance(od.decode(self.raw(DOCUMENT_KEY)), od.OffsetDocument)
def test_both_native_reset_confirmations_recover_corrupt_legacy_and_recheck_source(self):
from openpilot.starpilot.ui import feature_settings_compact as compact
from openpilot.starpilot.ui.runtime_app import StarShellSession
from openpilot.system.ui.widgets import DialogResult
class Dialog:
def __init__(self, _question, *_args, callback=None, **_kwargs):
self.confirmed = callback or _args[-1]
legacy = Path(self.params.get_param_path("Offset1"))
legacy.write_bytes(b"bad")
source = Path(self.params.get_param_path(DOCUMENT_KEY))
source.write_bytes(b"bad")
large = StarShellSession.__new__(StarShellSession)
cast(Any, large).feature_request = self.owner.apply
pushed = []
with patch("openpilot.system.ui.widgets.confirm_dialog.ConfirmDialog", Dialog), \
patch("openpilot.starpilot.ui.runtime_app.gui_app.push_widget", pushed.append):
large._confirm_feature_reset(self.row("slc_reset"))
legacy.write_bytes(b"changed")
source.write_bytes(b"changed")
pushed[-1].confirmed(DialogResult.CONFIRM)
self.assertIsNone(self.raw(DOCUMENT_KEY))
self.assertEqual(self.raw(DOCUMENT_KEY), b"changed")
large._confirm_feature_reset(self.row("slc_reset"))
pushed[-1].confirmed(DialogResult.CONFIRM)
self.assertFalse(read_params(self.params).enabled)
@@ -482,6 +388,7 @@ class SlcOffsetFeatureTests(unittest.TestCase):
Path(self.params.get_param_path(DOCUMENT_KEY)).unlink()
self.params.put_bool("SpeedLimitController", True, block=True)
source.write_bytes(b"bad")
compact_adapter = compact.FeatureSettingsCompact(cast(Any, NS(feature_request=self.owner.apply)))
pushed = []
with patch.object(compact, "BigConfirmationDialog", Dialog), \
@@ -508,7 +415,7 @@ class SlcOffsetFeatureTests(unittest.TestCase):
patch.object(compact, "BigConfirmationDialog", Dialog), \
patch("openpilot.starpilot.ui.runtime_app.gui_app.push_widget", pushed.append), \
patch.object(compact.gui_app, "texture", lambda *_args: None):
Path(self.params.get_param_path("Offset1")).write_bytes(b"bad")
Path(self.params.get_param_path(DOCUMENT_KEY)).write_bytes(b"bad")
recovery = self.row("slc_reset")
self.assertEqual(recovery.value, "Use default mph speed ranges")
large._confirm_feature_reset(recovery)
@@ -36,15 +36,12 @@ def published_cruise_mph(mph, *, software_target=True, metric=False):
return wire_kph * (1 / 3.6) / MPH
@pytest.mark.parametrize('document', [False, True])
@pytest.mark.parametrize('offset,delta', [(-25, 0), (-30, 0), (-24, 2), (-5, 0)])
def test_negative_offsets_cannot_remove_the_native_slc_ceiling(tmp_path, document, offset, delta):
def test_negative_offsets_cannot_remove_the_native_slc_ceiling(tmp_path, offset, delta):
drive = RetentionDrive(tmp_path)
params = Params(str(tmp_path))
if document:
params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (offset, 0, 0, 0, 0, 0, 0))), block=True)
else:
params.put('Offset1', float(offset), block=True)
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (offset * MPH, 0, 0, 0, 0, 0, 0))
params.put('SLCOffsetSchedule', od.to_value(document), block=True)
drive.runtime = Runtime(read_params(params))
drive.sm['carState'].vEgoCluster += delta
output, _, target = drive.step(0, limit=25)
@@ -71,7 +68,8 @@ def test_live_offset_edits_keep_the_accepted_limit_and_native_ceiling(tmp_path,
accepted = output.result.state.acceptance.accepted
session = runtime.session_id
for milliseconds, offset in ((1000, 5), (2000, -5)):
params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (0, 0, 0, offset, 0, 0, 0))), block=True)
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (0, 0, 0, offset * MPH, 0, 0, 0))
params.put('SLCOffsetSchedule', od.to_value(document), block=True)
receipt = object()
actions, cruise = [object()], [receipt]
refreshed, changed = lifecycle.refresh(NOW + milliseconds * 1_000_000, pending_actions=actions, pending_cruise=cruise)
@@ -101,7 +99,8 @@ def test_offset_edits_invalidate_old_prompts_without_accepting_a_new_limit(tmp_p
drive.step(50, request=Action(runtime.session_id, 1, first.message.slcState.decisionId, first.message.slcState.presentationId, 'accept'))
old, _, _ = drive.step(100, limit=50)
old_id = old.message.slcState.decisionId
params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (0, 0, 0, 5, 0, 0, 0))), block=True)
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (0, 0, 0, 5 * MPH, 0, 0, 0))
params.put('SLCOffsetSchedule', od.to_value(document), block=True)
assert lifecycle.refresh(NOW + 1_000_000_000)[0] is runtime
stale = Action(runtime.session_id, 2, old_id, old.message.slcState.presentationId, 'accept')
output, _, target = drive.step(1000, limit=50, request=stale)
@@ -125,7 +124,8 @@ def test_offset_edits_preserve_rejected_limits(tmp_path):
proposed, _, _ = drive.step(100, limit=50)
drive.step(150, limit=50, request=Action(runtime.session_id, 2, proposed.message.slcState.decisionId, proposed.message.slcState.presentationId, 'reject'))
rejected = runtime.state.acceptance.rejected
params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (0, 0, 0, 5, 0, 0, 0))), block=True)
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (0, 0, 0, 5 * MPH, 0, 0, 0))
params.put('SLCOffsetSchedule', od.to_value(document), block=True)
assert lifecycle.refresh(NOW + 1_000_000_000)[0] is runtime
output, _, target = drive.step(1000, limit=50)
assert not output.message.slcState.hasPending
@@ -203,7 +203,7 @@ def test_compact_rejection_or_timeout_keeps_set_speed_until_adoption(tmp_path, a
from openpilot.starpilot.ui.tests.test_slc_ui_actions import tap_sign, tap_control
drive = RetentionDrive(tmp_path, confirm=True)
params = Params(str(tmp_path))
params.put('SLCOffsetSchedule', od.to_value(od.adopt_legacy(False, (5.0,) * 7)), block=True)
params.put('SLCOffsetSchedule', od.to_value(od.OffsetDocument(od.IMPERIAL_BOUNDS, tuple(value * od.MPH_TO_MPS for value in (5.0,) * 7))), block=True)
drive.runtime = Runtime(read_params(params))
drive.ui.is_metric = metric
first, _, _ = drive.step(0)
@@ -192,9 +192,7 @@ class ValueEditorTests(unittest.TestCase):
def test_slc_steps_keep_direction_and_presets_keep_exact_source_guards(self):
self.session.feature_page = FeaturePage.SLC
adopt = next(row for row in self.session.feature_snapshot().rows if row.key == "slc_adopt")
self.assertTrue(self.owner.apply(replace(row_request(adopt, "confirm"), confirmation=True)))
editor = self.open_feature("Offset2")
editor = self.open_feature("slc:offset:2")
row = editor.rows[0]
self.assertEqual(value_request(editor, FeatureUiAction("change", row, 1)).direction, 1)
self.tap(editor, value_buttons(editor, row, feature_row_top(editor))[-1][1], self.session._feature_ui)