Restore Steering Settings

openpilot/starpilot/galaxy/settings.py

@@ -62,6 +62,7 @@ class AuthorityContext:
   cp: object | None
   cp_raw: bytes | None
   metric: bool = False
+  editing_saved_tune: bool = False
 
 
 class ContextSource(Protocol):
@@ -70,6 +71,8 @@ class ContextSource(Protocol):
 
 def _qualified(ctx: AuthorityContext, group: str) -> bool:
   cp = ctx.cp
+  if ctx.editing_saved_tune and group not in ("preferences", "parked_preferences", "torque"):
+    return False
   if group == "preferences":
     return True
   if group == "parked_preferences":
@@ -436,17 +439,29 @@ class SettingsGateway:
               "snapshot": [{"label": "Units", "value": "Metric" if ctx.metric else "Imperial"}],
               "note": "Saved configuration is shown here. It does not confirm that a feature is active while driving. Change settings in Toggles."}
 
+  def _context(self, page: str) -> AuthorityContext:
+    ctx = self.context.sample()
+    if page == "torque" and ctx.cp is None and ctx.parked and getattr(self.context, "offroad", lambda: False)():
+      from openpilot.starpilot.galaxy.vehicle_configuration import saved_torque_context
+      cp, token = saved_torque_context(self.params)
+      if cp is not None:
+        return replace(ctx, cp=cp, cp_raw=token, editing_saved_tune=True)
+    return ctx
+
+  def _fresh_context(self, ctx: AuthorityContext) -> AuthorityContext:
+    return self._context("torque") if ctx.editing_saved_tune else self.context.sample()
+
   def _owner(self, ctx: AuthorityContext, *, live: bool = False,
              session_valid: Callable[[], bool] = lambda: True) -> FeatureSettingsOwner:
     def current() -> AuthorityContext:
       if not session_valid():
         return AuthorityContext(False, None, None)
-      fresh = self.context.sample() if live else ctx
+      fresh = self._fresh_context(ctx) if live else ctx
       return fresh if fresh.cp_raw == ctx.cp_raw else AuthorityContext(False, None, fresh.cp_raw)
     def authorized(group: str) -> bool:
       if not session_valid():
         return False
-      fresh = self.context.sample() if live else ctx
+      fresh = self._fresh_context(ctx) if live else ctx
       return fresh.cp_raw == ctx.cp_raw and _qualified(fresh, group)
     return FeatureSettingsOwner(self.params, authorized,
                                 vehicle_fingerprint=lambda: getattr(current().cp, "carFingerprint", None),
@@ -521,10 +536,16 @@ class SettingsGateway:
       return self._sentry_owner(ctx).snapshot()
     if page == "vasm":
       return self._vasm_owner(ctx).snapshot()
-    return self._owner(ctx).snapshot(page, parked=ctx.parked,
-                                     system_long=_qualified(ctx, "long"),
-                                     lateral_context=_qualified(ctx, "lane") or (page == "lane" and _qualified(ctx, "lane_live")),
-                                     metric=ctx.metric, configure_while_driving=True)
+    state = self._owner(ctx).snapshot(page, parked=ctx.parked,
+                                      system_long=_qualified(ctx, "long"),
+                                      lateral_context=_qualified(ctx, "lane") or (page == "lane" and _qualified(ctx, "lane_live")),
+                                      metric=ctx.metric, configure_while_driving=True)
+    if ctx.editing_saved_tune:
+      from opendbc.car.values import PLATFORMS
+      platform = PLATFORMS[ctx.cp.carFingerprint]
+      label = platform.config.car_docs[0].name if platform.config.car_docs else str(ctx.cp.carFingerprint).replace("_", " ").title()
+      return replace(state, subtitle=f"{label} · Saved steering settings")
+    return state
 
   def _clean(self) -> None:
     now = self.clock()
@@ -534,7 +555,7 @@ class SettingsGateway:
   def page(self, page: str, token: str, generation: bytes) -> dict:
     if page not in PAGES:
       raise SettingsUnavailable("Unknown page")
-    ctx = self.context.sample()
+    ctx = self._context(page)
     state = self._state(page, ctx)
     editor = None
     if page == "vasm":
@@ -574,7 +595,7 @@ class SettingsGateway:
     if view is None or view.session != _session_key(token) or view.generation != generation or \
        type(index) is not int or not 0 <= index < len(view.rows) or type(direction) is not int or direction not in (-1, 0, 1):
       raise SettingsChanged("Refresh settings")
-    ctx = self.context.sample()
+    ctx = self._context(view.page)
     if ctx.cp_raw != view.cp_raw:
       raise SettingsChanged("Vehicle context changed")
     state = self._state(view.page, ctx)
@@ -661,7 +682,7 @@ class SettingsGateway:
       if intent is None or intent.session != _session_key(token) or intent.generation != generation:
         raise SettingsChanged("Confirmation expired")
       del self.intents[intent_id]
-    ctx = self.context.sample()
+    ctx = self._context(intent.page)
     if not session_valid() or not (ctx.parked or _onroad_preference(intent.page, intent.request.key)) or ctx.cp_raw != intent.cp_raw:
       raise SettingsChanged("Vehicle changed or this operation requires offroad mode")
     if intent.page == "pip":

openpilot/starpilot/galaxy/tests/test_saved_torque_configuration.py

@@ -0,0 +1,72 @@
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+from openpilot.common.params import Params
+from openpilot.starpilot.galaxy.vehicle_configuration import saved_torque_context
+from openpilot.starpilot.lateral.controller_selection import KNOWN_POLICIES
+from openpilot.starpilot.lateral.torque_settings import FieldChoice, PlatformProfile, serialize_document
+from openpilot.starpilot.vehicle_selection import encode
+
+
+class TestSavedTorqueConfiguration(unittest.TestCase):
+  def setUp(self):
+    temporary = tempfile.TemporaryDirectory()
+    self.addCleanup(temporary.cleanup)
+    self.params = Params(temporary.name)
+    self.params.put('LateralControllerSelection', json.loads(
+      '{"vehicles":{"HYUNDAI_IONIQ_6":{"brand":"hyundai","mode":"starpilot"}},"version":1}'), block=True)
+    self.params.put('TorqueOverrideDocument', json.loads(
+      '{"schemaVersion":1,"vehicles":{"HYUNDAI_IONIQ_6":{"basis":{"friction":0.09000000357627869,' +
+      '"latAccelFactor":3.0,"latAccelOffset":0.0},"factor":{"customValue":2.1,"mode":"source"},' +
+      '"friction":{"customValue":null,"mode":"source"}}}}'), block=True)
+
+  def save(self, identities, factor=1.8, mode='starpilot'):
+    controller = {'version': 1, 'vehicles': {identity: {'brand': KNOWN_POLICIES[identity], 'mode': mode} for identity in identities}}
+    self.params.put('LateralControllerSelection', controller, block=True)
+    document = serialize_document({identity: PlatformProfile((1.8, 0.0, 0.1), FieldChoice('custom', factor), FieldChoice())
+                                   for identity in identities})
+    self.params.put('TorqueOverrideDocument', json.loads(document), block=True)
+
+  def test_retained_single_profile_can_edit_without_current_vehicle(self):
+    cp, token = saved_torque_context(self.params)
+    self.assertEqual(cp.carFingerprint, 'HYUNDAI_IONIQ_6')
+    self.assertTrue(token.startswith(b'saved-torque-configuration:'))
+    self.assertFalse(Path(self.params.get_param_path('VehicleSelection')).exists())
+    for key in ('CarParams', 'CarParamsPersistent', 'CarParamsCache'):
+      self.assertFalse(Path(self.params.get_param_path(key)).exists())
+
+  def test_normal_profile_edits_do_not_invalidate_configuration_token(self):
+    _, token = saved_torque_context(self.params)
+    self.save(('HYUNDAI_IONIQ_6',), factor=2.0, mode='standard')
+    self.assertEqual(saved_torque_context(self.params)[1], token)
+    self.params.put_bool('AlphaLongitudinalEnabled', True, block=True)
+    self.assertNotEqual(saved_torque_context(self.params)[1], token)
+
+  def test_ambiguous_or_malformed_saved_documents_fail_closed(self):
+    self.save(('HYUNDAI_IONIQ_6', 'CHEVROLET_BOLT_CC_2018_2021'))
+    self.assertEqual(saved_torque_context(self.params), (None, None))
+    for key in ('TorqueOverrideDocument', 'LateralControllerSelection'):
+      self.save(('HYUNDAI_IONIQ_6',))
+      self.params.put(key, {}, block=True)
+      self.assertEqual(saved_torque_context(self.params), (None, None))
+
+  def test_unreadable_missing_or_manual_selected_are_not_fallback_contexts(self):
+    self.params.put('VehicleSelection', json.loads(encode('HYUNDAI_IONIQ_6')), block=True)
+    self.assertEqual(saved_torque_context(self.params), (None, None))
+    self.params.put('VehicleSelection', json.loads(encode(None)), block=True)
+    self.params.remove('TorqueOverrideDocument')
+    self.assertEqual(saved_torque_context(self.params), (None, None))
+    self.save(('HYUNDAI_IONIQ_6',))
+    path = Path(self.params.get_param_path('TorqueOverrideDocument'))
+    path.unlink()
+    path.symlink_to('LateralControllerSelection')
+    self.assertEqual(saved_torque_context(self.params), (None, None))
+
+  def test_each_document_must_name_the_same_single_vehicle(self):
+    self.params.put('LateralControllerSelection', {'version': 1, 'vehicles': {}}, block=True)
+    self.assertEqual(saved_torque_context(self.params), (None, None))
+    self.save(('HYUNDAI_IONIQ_6',))
+    self.params.put('TorqueOverrideDocument', {'schemaVersion': 1, 'vehicles': {}}, block=True)
+    self.assertEqual(saved_torque_context(self.params), (None, None))

openpilot/starpilot/galaxy/tests/test_saved_torque_page.py

@@ -0,0 +1,68 @@
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from openpilot.common.params import Params
+from openpilot.starpilot.galaxy.settings import LiveContextSource, SettingsChanged, SettingsGateway
+
+
+class TestSavedTorquePage(unittest.TestCase):
+  def setUp(self):
+    temporary = tempfile.TemporaryDirectory()
+    self.addCleanup(temporary.cleanup)
+    self.params = Params(temporary.name)
+    self.params.put('LateralControllerSelection', {'vehicles': {'HYUNDAI_IONIQ_6': {'brand': 'hyundai', 'mode': 'starpilot'}}, 'version': 1}, block=True)
+    self.params.put('TorqueOverrideDocument', {'schemaVersion': 1, 'vehicles': {'HYUNDAI_IONIQ_6': {
+      'basis': {'friction': 0.09000000357627869, 'latAccelFactor': 3.0, 'latAccelOffset': 0.0},
+      'factor': {'customValue': 2.1, 'mode': 'source'}, 'friction': {'customValue': None, 'mode': 'source'},
+    }}}, block=True)
+    self.params.put_bool('OpenpilotEnabledToggle', True, block=True)
+    self.params.put_bool('AlphaLongitudinalEnabled', True, block=True)
+    self.params.put_bool('ForceAutoTuneOff', True, block=True)
+    self.context = LiveContextSource(self.params)
+    self.gateway = SettingsGateway(self.params, self.context)
+    self.addCleanup(self.gateway.close)
+    offroad = patch.object(self.context, 'offroad', return_value=True)
+    self.offroad = offroad.start()
+    self.addCleanup(offroad.stop)
+
+  def page(self, name='torque'):
+    return self.gateway.page(name, 'session', b'generation')
+
+  def test_retained_tune_restores_full_flat_page_and_can_save(self):
+    page = self.page()
+    rows = {row['label']: row for row in page['rows']}
+    self.assertEqual(set(rows), {'Steering Controller', 'Turn Assist', 'Prep My Vehicle for Tuning',
+                                'Automatic Steering Learning', 'Lat Accel', 'Lateral acceleration — Reset to Default',
+                                'Friction', 'Friction — Reset to Default'})
+    self.assertIn('Saved steering settings', page['subtitle'])
+    self.assertFalse(any(row['page'] for row in page['rows']))
+    for label in ('Steering Controller', 'Turn Assist', 'Prep My Vehicle for Tuning', 'Lat Accel', 'Friction'):
+      self.assertTrue(rows[label]['available'], label)
+    self.assertFalse(rows['Automatic Steering Learning']['available'])
+    index = next(i for i, row in enumerate(page['rows']) if row['label'] == 'Friction')
+    intent = self.gateway.preview(page['view'], index, 1, 'session', b'generation')
+    self.assertTrue(self.gateway.confirm(intent['intent'], 'session', b'generation'))
+    self.assertAlmostEqual(self.params.get('TorqueOverrideDocument')['vehicles']['HYUNDAI_IONIQ_6']['friction']['customValue'], .10)
+    for key in ('VehicleSelection', 'CarParams', 'CarParamsPersistent', 'CarParamsCache'):
+      self.assertFalse(Path(self.params.get_param_path(key)).exists(), key)
+
+  def test_saved_tune_does_not_supply_other_pages_or_live_context(self):
+    self.assertTrue(self.gateway._context('torque').editing_saved_tune)
+    self.assertIsNone(self.context.sample().cp)
+    self.assertIsNone(self.gateway._context('slc').cp)
+    slc = next(row for row in self.page('slc')['rows'] if row['label'] == 'Speed Limit Controller')
+    self.assertFalse(slc['available'])
+    self.offroad.return_value = False
+    with patch.object(self.context, 'configuration_allowed', return_value=True):
+      self.assertIsNone(self.gateway._context('torque').cp)
+
+  def test_pending_edit_is_revoked_when_offroad_context_ends(self):
+    page = self.page()
+    index = next(i for i, row in enumerate(page['rows']) if row['label'] == 'Friction')
+    intent = self.gateway.preview(page['view'], index, 1, 'session', b'generation')
+    self.offroad.return_value = False
+    with patch.object(self.context, 'configuration_allowed', return_value=True), self.assertRaises(SettingsChanged):
+      self.gateway.confirm(intent['intent'], 'session', b'generation')
+    self.assertEqual(self.params.get('TorqueOverrideDocument')['vehicles']['HYUNDAI_IONIQ_6']['friction']['mode'], 'source')

openpilot/starpilot/galaxy/vehicle_configuration.py

@@ -19,6 +19,17 @@ def configuration_context(params, physical_cp, physical_raw):
   revision = hashlib.sha256(repr((selection.raw, saved)).encode()).digest()
   if physical_cp is not None and physical_cp.carFingerprint == selection.platform:
     return physical_cp, b'vehicle-configuration:' + revision + (physical_raw or b'')
+  cp, raw = _factory_context(params, selection.platform, saved)
+  if read_selection(params) != selection or _startup_snapshot(params) != saved:
+    return None, None
+  return (cp, b'vehicle-configuration:' + revision + raw) if cp is not None else (None, None)
+
+
+def _startup_snapshot(params):
+  return tuple((key, *read_saved(params, key, 512)) for key in STARTUP_KEYS)
+
+
+def _factory_context(params, platform, saved):
   try:
     from opendbc.car import gen_empty_fingerprint
     from opendbc.car.car_helpers import interfaces
@@ -28,12 +39,46 @@ def configuration_context(params, physical_cp, physical_raw):
     values = {key: (raw, readable) for key, raw, readable in saved}
     release = values['IsReleaseBranch'] == (b'1', True)
     alpha = not release and values['AlphaLongitudinalEnabled'] == (b'1', True)
-    cp = interfaces[selection.platform].get_params(selection.platform, gen_empty_fingerprint(), [], alpha, release, False)
+    cp = interfaces[platform].get_params(platform, gen_empty_fingerprint(), [], alpha, release, False)
     VehicleStartupPreferences.read(params, enabled=values['OpenpilotEnabledToggle'] == (b'1', True)).prepare(cp)
-    if read_selection(params) != selection or tuple((key, *read_saved(params, key, 512)) for key in STARTUP_KEYS) != saved:
-      return None, None
     raw = cp.to_bytes()
     reader = messaging.log_from_bytes(raw, car.CarParams)
-    return reader, b'vehicle-configuration:' + revision + raw
+    return reader, raw
+  except (AttributeError, KeyError, OSError, RuntimeError, TypeError, ValueError, OverflowError):
+    return None, None
+
+
+def _saved_torque_identities(params):
+  from openpilot.starpilot.lateral import controller_selection, torque_settings
+  controller_raw, controller_readable = read_saved(params, controller_selection.DOCUMENT_KEY,
+                                                   controller_selection.MAX_DOCUMENT_BYTES)
+  torque_raw, torque_readable = read_saved(params, torque_settings.DOCUMENT_KEY, torque_settings.MAX_DOCUMENT_BYTES)
+  if not controller_readable or not torque_readable or controller_raw is None or torque_raw is None:
+    raise ValueError('Saved torque profiles unavailable')
+  controller = controller_selection.parse_document(controller_raw)
+  torque = torque_settings.parse_document(torque_raw)
+  controller_ids = set(controller['vehicles'])
+  torque_ids = set(torque)
+  identities = controller_ids | torque_ids
+  if len(identities) != 1 or controller_ids != torque_ids:
+    raise ValueError('Saved torque vehicle is ambiguous')
+  return tuple(sorted(identities))
+
+
+def saved_torque_context(params):
+  selection = read_selection(params)
+  if not selection.readable or not selection.valid or selection.platform is not None:
+    return None, None
+  try:
+    identities = _saved_torque_identities(params)
+    saved = _startup_snapshot(params)
+    cp, raw = _factory_context(params, identities[0], saved)
+    from openpilot.starpilot.lateral.torque_runtime import production_supported_cp
+    if cp is None or not production_supported_cp(cp):
+      return None, None
+    if read_selection(params) != selection or _startup_snapshot(params) != saved or _saved_torque_identities(params) != identities:
+      return None, None
+    revision = hashlib.sha256(repr((selection.raw, saved, identities)).encode()).digest()
+    return cp, b'saved-torque-configuration:' + revision + raw
   except (AttributeError, KeyError, OSError, RuntimeError, TypeError, ValueError, OverflowError):
     return None, None

openpilot/starpilot/ui/torque_feature.py

@@ -10,7 +10,7 @@ from openpilot.starpilot.lateral.torque_settings import (
   DOCUMENT_KEY, LEGACY_KEYS, FieldChoice, LegacyMode, PlatformProfile, bounds as torque_bounds,
   interpret_legacy, parse_document, replace_field, resolve_document, serialize_document,
 )
-from openpilot.starpilot.ui.feature_settings_state import FeaturePage, FeatureRow, FeatureSettingsRequest, TORQUE_CONFIRM_ACTIONS
+from openpilot.starpilot.ui.feature_settings_state import FeatureRow, FeatureSettingsRequest, TORQUE_CONFIRM_ACTIONS
 
 TORQUE_NUMBERS = LEGACY_KEYS
 
@@ -57,8 +57,7 @@ class TorqueFeature:
     bolt = capability is not None and capability[0] in BOLT_VEHICLES
     try:
       if capability is None:
-        return False, (FeatureRow("", "Torque profile", "Select a supported vehicle in Vehicle settings",
-                                  page=FeaturePage.VEHICLE, available=True),)
+        return False, (FeatureRow("", "Vehicle", "Select your vehicle in Vehicle settings to configure steering"),)
       fingerprint = capability[0]
       basis = (capability[5], capability[6], capability[7])
       if raw is None: