RAV4 torque tuning

openpilot/selfdrive/controls/lib/latcontrol_torque.py

@@ -74,6 +74,11 @@ class LatControlTorque(LatControl):
     if self.starpilot_extension is not None:
       return self.starpilot_extension.update(active, CS, VM, params, steer_limited_by_safety,
                                                         desired_curvature, curvature_limited, lat_delay)
+    return self.update_standard(active, CS, VM, params, steer_limited_by_safety,
+                                desired_curvature, curvature_limited, lat_delay)
+
+  def update_standard(self, active, CS, VM, params, steer_limited_by_safety, desired_curvature, curvature_limited, lat_delay,
+                      *, friction_policy=None):
     pid_log = log.ControlsState.LateralTorqueState.new_message()
     pid_log.version = VERSION
     measured_curvature = -VM.calc_curvature(math.radians(CS.steeringAngleDeg - params.angleOffsetDeg), CS.vEgo, params.roll)
@@ -99,6 +104,8 @@ class LatControlTorque(LatControl):
     ff -= self.torque_params.latAccelOffset
     threshold = flm.base_threshold(self.flm_surface, CS.vEgo, FRICTION_THRESHOLD)
     ff, threshold = flm.stages(self.flm_surface, CS, setpoint, desired_lateral_jerk, ff, threshold)
+    if friction_policy is not None:
+      threshold = friction_policy.friction_threshold(CS.vEgo, setpoint, desired_lateral_jerk)
     ff += get_friction(error + JERK_GAIN * desired_lateral_jerk, lateral_accel_deadzone, threshold, self.torque_params)
 
     if not active:
@@ -115,6 +122,9 @@ class LatControlTorque(LatControl):
       output_torque = apply_turn_assist(self, CS, VM, params, desired_curvature, output_torque)
       output_torque = flm.angle_assist(self.flm_surface, CS, VM, params, desired_curvature, setpoint, desired_lateral_jerk, output_torque)
 
+      if friction_policy is not None:
+        output_torque *= friction_policy.output_scale(setpoint, CS.vEgo)
+
       pid_log.active = True
       pid_log.p = float(self.pid.p)
       pid_log.i = float(self.pid.i)

openpilot/starpilot/lateral/controller_selection.py

@@ -16,7 +16,7 @@ DOCUMENT_VERSION = 1
 LEARNING_OFF_KEY = 'ForceAutoTuneOff'
 MAX_DOCUMENT_BYTES = 4096
 KNOWN_POLICIES = {'HYUNDAI_IONIQ_6': 'hyundai', 'GENESIS_G70_2020': 'hyundai', 'GENESIS_GV70_ELECTRIFIED_1ST_GEN': 'hyundai', 'TOYOTA_COROLLA_TSS2': 'toyota'}
-KNOWN_POLICIES.update({'HYUNDAI_SONATA_HYBRID': 'hyundai'})
+KNOWN_POLICIES.update({'HYUNDAI_SONATA_HYBRID': 'hyundai', 'TOYOTA_RAV4_TSS2': 'toyota'})
 KNOWN_POLICIES.update(dict.fromkeys(('HYUNDAI_PALISADE', 'HYUNDAI_PALISADE_2023'), 'hyundai'))
 KNOWN_POLICIES.update(dict.fromkeys((
   'CHEVROLET_BOLT_CC_2017', 'CHEVROLET_BOLT_CC_2018_2021', 'CHEVROLET_BOLT_CC_2022_2023',
@@ -116,6 +116,9 @@ def policy_for(CP) -> str | None:
     return 'bolt'
   if corolla_supported(CP):
     return 'corolla_tss2'
+  from openpilot.starpilot.lateral.rav4_tss2_policy import supported_cp as rav4_supported
+  if rav4_supported(CP):
+    return 'rav4_tss2'
   from openpilot.starpilot.lateral.palisade_policy import supported_cp as palisade_supported
   from openpilot.starpilot.lateral.sonata_hybrid_policy import supported_cp as sonata_hybrid_supported
   if sonata_hybrid_supported(CP):
@@ -223,6 +226,11 @@ def read_selection(params, CP) -> ControllerSelection:
     return ControllerSelection(default.mode, default.policy, 'invalid')
 
 
+def controller_uses_native_learning(selection: ControllerSelection) -> bool:
+  # RAV4 adds only two shaping stages to the modern native controller.
+  return selection.mode == ControllerMode.STANDARD or selection.policy == "rav4_tss2"
+
+
 def learning_allowed(params, CP, *, selection: ControllerSelection | None = None) -> bool:
   """Latch the exact controller's learning policy at each process startup.
 
@@ -241,4 +249,4 @@ def learning_allowed(params, CP, *, selection: ControllerSelection | None = None
     return False
   if geometry.force_auto:
     return selected.source != 'invalid'
-  return selected.mode != ControllerMode.STARPILOT
+  return controller_uses_native_learning(selected) and (selected.policy != "rav4_tss2" or selected.source != "invalid")

openpilot/starpilot/lateral/rav4_tss2_policy.py

@@ -0,0 +1,47 @@
+"""RAV4 TSS2 friction/taper stages on the modern torque controller."""
+import math
+
+import numpy as np
+
+from opendbc.car import structs
+from opendbc.car.toyota.values import CAR
+from openpilot.starpilot.lateral.torque_shaping import get_standard_friction_threshold, sigmoid
+
+
+def supported_cp(cp):
+  if (cp.brand != "toyota" or cp.carFingerprint != CAR.TOYOTA_RAV4_TSS2 or cp.notCar or cp.passive or cp.dashcamOnly or
+      cp.steerControlType != structs.CarParams.SteerControlType.torque or cp.lateralTuning.which() != "torque"):
+    return False
+  tune = cp.lateralTuning.torque
+  return (math.isfinite(tune.latAccelFactor) and tune.latAccelFactor > 0 and
+          math.isfinite(tune.latAccelOffset) and math.isfinite(tune.friction) and tune.friction >= 0)
+
+
+def center_envelope(setpoint, speed):
+  return sigmoid((13.0 - max(speed, 0.0)) / 3.0) * sigmoid((0.30 - abs(setpoint)) / 0.08)
+
+
+def friction_threshold(speed, setpoint, jerk):
+  base = get_standard_friction_threshold(speed) * (1.0 + .14 * center_envelope(setpoint, speed))
+  curve = float(np.interp(abs(setpoint), [.30, .85], [1.0, 0.0]))
+  calm = float(np.interp(abs(jerk), [.25, .75], [1.0, 0.0]))
+  increment = float(np.interp(max(speed, 0.0), [0., 2.5, 5., 10., 15., 25.], [0., 0., .04, .065, .05, .025]))
+  return base + increment * curve * calm
+
+
+def center_output_scale(setpoint, speed):
+  return 1.0 - .12 * center_envelope(setpoint, speed)
+
+
+class Rav4TSS2TorquePolicy:
+  def __init__(self, parent, cp):
+    if not supported_cp(cp):
+      raise ValueError("Unsupported RAV4 TSS2 torque profile")
+    self.parent = parent
+
+  friction_threshold = staticmethod(friction_threshold)
+  output_scale = staticmethod(center_output_scale)
+
+  def update(self, active, cs, vm, params, safety_limited, curvature, curvature_limited, delay):
+    return self.parent.update_standard(active, cs, vm, params, safety_limited, curvature, curvature_limited, delay,
+                                       friction_policy=self)

openpilot/starpilot/lateral/tests/test_controller_selection.py

@@ -55,7 +55,7 @@ class TestControllerSelection(unittest.TestCase):
                 (TOYOTA.TOYOTA_COROLLA_TSS2, 'corolla_tss2'),
                 (HYUNDAI.KIA_EV6, 'kia_ev6'),
                 (HYUNDAI.HYUNDAI_SONATA, 'sonata'),
-                (TOYOTA.TOYOTA_RAV4_TSS2, None))
+                (TOYOTA.TOYOTA_RAV4_TSS2, 'rav4_tss2'))
     for vehicle, policy in expected:
       with self.subTest(vehicle=vehicle):
         cp = cp_for(vehicle)
@@ -96,8 +96,8 @@ class TestControllerSelection(unittest.TestCase):
     wrong.notCar = True
     self.assertIsNone(policy_for(wrong))
     with self.assertRaises(ValueError):
-      unsupported = cp_for(TOYOTA.TOYOTA_RAV4_TSS2)
-      LatControlTorque(unsupported.as_reader(), interfaces[TOYOTA.TOYOTA_RAV4_TSS2](unsupported), DT_CTRL,
+      unsupported = cp_for(TOYOTA.TOYOTA_RAV4_TSS2_2022)
+      LatControlTorque(unsupported.as_reader(), interfaces[TOYOTA.TOYOTA_RAV4_TSS2_2022](unsupported), DT_CTRL,
                        controller_mode=ControllerMode.STARPILOT)
 
   def test_standard_skips_policy_constructor_and_ioniq_factor_multiplier(self):

openpilot/starpilot/lateral/tests/test_corolla_tss2_policy.py

@@ -57,7 +57,7 @@ def test_exact_vehicle_admission_and_upstream_gains():
     assert native.pid._k_i[1] == [0.15]
   other_cp, other, _ = controller_for(CAR.TOYOTA_RAV4_TSS2)
   assert not supported_cp(other_cp)
-  assert selected_policy(other) is None
+  assert type(selected_policy(other)).__name__ == "Rav4TSS2TorquePolicy"
   assert other.pid._k_i[1] == [0.15]
 
 

openpilot/starpilot/lateral/tests/test_rav4_tss2_policy.py

@@ -0,0 +1,235 @@
+import json
+import math
+import os
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from opendbc.car import structs
+from opendbc.car.car_helpers import interfaces
+from opendbc.car.lateral import get_friction
+from opendbc.car.toyota.values import CAR
+from opendbc.car.vehicle_model import VehicleModel
+from openpilot.cereal import log
+from openpilot.common.params import Params
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.selfdrive.controls.controlsd import Controls
+from openpilot.selfdrive.controls.lib.latcontrol_torque import LatControlTorque
+from openpilot.starpilot.lateral.controller_selection import (
+  ControllerMode, DOCUMENT_KEY, LEARNING_OFF_KEY, default_selection, policy_for, replace_mode,
+)
+from openpilot.starpilot.lateral.rav4_tss2_policy import (
+  supported_cp, friction_threshold, center_output_scale,
+)
+from openpilot.starpilot.lateral.tests.test_lane_runtime import feed
+from openpilot.starpilot.lateral.tests.test_torque_learning import learned_event
+from openpilot.starpilot.lateral.torque_extension import selected_policy
+from openpilot.starpilot.ui.feature_settings_owner import FeatureSettingsOwner
+from openpilot.starpilot.ui.feature_settings_state import row_change
+
+
+def raw_saved(params, key):
+  try:
+    return Path(params.get_param_path(key)).read_bytes()
+  except FileNotFoundError:
+    return None
+
+
+def factory(car=CAR.TOYOTA_RAV4_TSS2):
+  return interfaces[car].get_non_essential_params(car)
+
+
+def publish_rav4_params(params):
+  """Publish RAV4 CP through Card's real startup configuration and finalization."""
+  from openpilot.selfdrive.car.card import Car
+  from openpilot.starpilot.car.toyota.aol import qualified
+  params.put_bool("OpenpilotEnabledToggle", True, block=True)
+  params.put_bool("SafeMode", False, block=True)
+  cp = factory()
+
+  def discover(*args, pre_create_hook, **kwargs):
+    prepared = pre_create_hook(cp.as_reader().as_builder(), cp.carFingerprint, {}, [])
+    return interfaces[cp.carFingerprint](prepared)
+
+  with patch.dict(os.environ, {"SIMULATION": "1"}), \
+       patch("openpilot.selfdrive.car.card.messaging.recv_one_retry", return_value=SimpleNamespace(can=[1])), \
+       patch("openpilot.selfdrive.car.card.get_car", side_effect=discover):
+    selected = Car()
+  try:
+    final = selected.CP.as_reader().as_builder()
+    assert qualified(final, marked_only=True)
+    assert final.lateralTuning.torque.to_dict() == cp.lateralTuning.torque.to_dict()
+    for field in ("carFingerprint", "steerControlType", "mass", "wheelbase", "centerToFront", "steerRatio",
+                  "tireStiffnessFront", "tireStiffnessRear", "steerActuatorDelay"):
+      assert getattr(final, field) == getattr(cp, field)
+    assert raw_saved(params, "CarParams") == final.to_bytes()
+    return final
+  finally:
+    selected.vehicle_startup.close()
+
+
+def controller(cp, mode=None):
+  return LatControlTorque(cp.as_reader(), interfaces[cp.carFingerprint](cp), .01, controller_mode=mode)
+
+
+@pytest.mark.parametrize("speed,increment", [(0., 0.), (2.5, 0.), (5., .04), (10., .065), (15., .05), (25., .025)])
+def test_latest_cleanup_knots_taper_and_left_right_symmetry(speed, increment):
+  # Frozen source-default center envelope, with both taper plateaus independently exercised.
+  envelope = (1 / (1 + math.exp((speed - 13) / 3))) * (1 / (1 + math.exp(-.30 / .08)))
+  assert friction_threshold(speed, 0, 0) == pytest.approx(.30 * (1 + .14 * envelope) + increment)
+  for accel, jerk in ((.10, .10), (.40, .40), (.85, .75)):
+    assert friction_threshold(speed, accel, jerk) == friction_threshold(speed, -accel, -jerk)
+  for accel, jerk in ((.85, 0), (0, .75)):
+    env = (1 / (1 + math.exp((speed - 13) / 3))) * (1 / (1 + math.exp((abs(accel) - .30) / .08)))
+    assert friction_threshold(speed, accel, jerk) == pytest.approx(.30 * (1 + .14 * env))
+  assert .88 <= center_output_scale(0, speed) <= 1
+
+
+def test_exact_scope_excludes_related_racks_and_bad_tunes():
+  cp = factory()
+  assert supported_cp(cp)
+  assert policy_for(cp) == "rav4_tss2"
+  assert default_selection(cp).mode == ControllerMode.STARPILOT
+  for car in (CAR.TOYOTA_RAV4_TSS2_2022, CAR.TOYOTA_RAV4_TSS2_2023, CAR.TOYOTA_RAV4_PRIME, CAR.TOYOTA_RAV4H):
+    assert not supported_cp(factory(car))
+  for field, value in (("brand", "hyundai"), ("notCar", True), ("passive", True), ("dashcamOnly", True),
+                       ("steerControlType", structs.CarParams.SteerControlType.angle)):
+    bad = cp.as_reader().as_builder()
+    setattr(bad, field, value)
+    assert not supported_cp(bad)
+  for field, value in (("latAccelFactor", 0), ("latAccelFactor", float("nan")), ("friction", -.1),
+                       ("latAccelOffset", float("inf"))):
+    bad = cp.as_reader().as_builder()
+    setattr(bad.lateralTuning.torque, field, value)
+    assert not supported_cp(bad)
+
+
+def test_actual_modern_owner_reaches_friction_and_final_taper_preserves_tune():
+  cp = factory()
+  shaped = controller(cp)
+  assert type(selected_policy(shaped)).__name__ == "Rav4TSS2TorquePolicy"
+  assert shaped.pid._k_i[1] == [.15]  # Modern control gains stay at the existing value.
+  shaped.update_torque_parameters(3.2, .025, .08)
+  assert shaped.torque_params.latAccelFactor == pytest.approx(3.2)
+  assert shaped.torque_params.latAccelOffset == pytest.approx(.025)
+  assert shaped.torque_params.friction == pytest.approx(.08)
+  cs = SimpleNamespace(vEgo=10., steeringAngleDeg=0., steeringPressed=False)
+  params = log.VehicleParameters.new_message(angleOffsetDeg=0., roll=0.)
+  vm = VehicleModel(cp)
+  with patch("openpilot.selfdrive.controls.lib.latcontrol_torque.get_friction", wraps=get_friction) as friction, \
+       patch.object(shaped.pid, "update", return_value=.4):
+    for _ in range(300):
+      result, _, state = shaped.update(True, cs, vm, params, False, .001, False, .14)
+    assert state.desiredLateralAccel == pytest.approx(.1)
+    assert state.desiredLateralJerk == pytest.approx(0., abs=1e-8)
+    assert friction.call_args.args[2] == pytest.approx(friction_threshold(10., .1, 0.), abs=1e-8)
+    raw = shaped.torque_from_lateral_accel(.4, shaped.torque_params)
+    assert result == pytest.approx(-raw * center_output_scale(.1, 10.))
+  standard = controller(cp, ControllerMode.STANDARD)
+  assert selected_policy(standard) is None
+  with patch("openpilot.selfdrive.controls.lib.latcontrol_torque.get_friction", wraps=get_friction) as friction, \
+       patch.object(standard.pid, "update", return_value=.4):
+    standard.update(True, cs, vm, params, False, .001, False, .14)
+    assert friction.call_args.args[2] == .20
+
+
+def test_actual_controls_saved_optout_and_native_learning_consumer():
+  with OpenpilotPrefix(), patch.dict(os.environ, {"SIMULATION": "0", "REPLAY": "1", "TORQUE_REPLAY_RUNTIME": "0",
+                                                "AOL_REPLAY_RUNTIME": "0", "LANE_CENTERING_REPLAY_RUNTIME": "0"}), \
+       patch("openpilot.selfdrive.controls.controlsd.messaging.PubMaster"):
+    saved = Params()
+    cp = publish_rav4_params(saved)
+    saved.put_bool("OpenpilotEnabledToggle", True, block=True)
+    saved.put_bool("SafeMode", False, block=True)
+    shaped = Controls()
+    assert shaped.LaC.controller_mode == ControllerMode.STARPILOT
+    assert shaped.torque_learning_allowed
+    for tick in range(2):
+      now = 1_000_000_000 + tick * 250_000_000
+      feed(shaped, now, 0)
+      shaped.sm.update_msgs(now / 1e9, [learned_event(cp, now)])
+    assert shaped.sm.all_checks(["lateralTorqueParameters"])
+    with patch.object(shaped.LaC, "update_torque_parameters") as update:
+      command, _ = shaped.state_control()
+      assert command.latActive
+      update.assert_called_once()
+    raw = replace_mode(None, cp, ControllerMode.STANDARD)
+    saved.put(DOCUMENT_KEY, json.loads(raw), block=True)
+    stored_raw = raw_saved(saved, DOCUMENT_KEY)
+    assert json.loads(stored_raw) == json.loads(raw)
+    assert shaped.LaC.controller_mode == ControllerMode.STARPILOT
+    standard = Controls()
+    assert standard.LaC.controller_mode == ControllerMode.STANDARD
+    assert selected_policy(standard.LaC) is None
+    assert raw_saved(saved, DOCUMENT_KEY) == stored_raw
+
+
+@pytest.mark.parametrize("mode", [ControllerMode.STANDARD, ControllerMode.STARPILOT])
+@pytest.mark.parametrize("saved_raw,allowed,label", [(None, True, "On"), (b"0", True, "On"),
+                                                      (b"1", False, "Off"), (b"bad", False, "Invalid saved preference")])
+def test_actual_controls_consumer_and_ui_row_honor_learning_choice(mode, saved_raw, allowed, label):
+  with OpenpilotPrefix(), patch.dict(os.environ, {"SIMULATION": "0", "REPLAY": "1", "TORQUE_REPLAY_RUNTIME": "0",
+                                                "AOL_REPLAY_RUNTIME": "0", "LANE_CENTERING_REPLAY_RUNTIME": "0"}), \
+       patch("openpilot.selfdrive.controls.controlsd.messaging.PubMaster"):
+    saved = Params()
+    cp = publish_rav4_params(saved)
+    saved.put(DOCUMENT_KEY, json.loads(replace_mode(None, cp, mode)), block=True)
+    saved.put_bool("OpenpilotEnabledToggle", True, block=True)
+    saved.put_bool("SafeMode", False, block=True)
+    if saved_raw is not None:
+      Path(saved.get_param_path(LEARNING_OFF_KEY)).write_bytes(saved_raw)
+    controls = Controls()
+    assert controls.LaC.controller_mode == mode
+    assert controls.torque_learning_allowed == allowed
+    owner = FeatureSettingsOwner(saved, lambda group: group == "preferences",
+                                 vehicle_fingerprint=lambda: str(cp.carFingerprint), vehicle_params=lambda: cp)
+    snapshot = owner.snapshot("torque", parked=False, system_long=False, lateral_context=False, metric=False)
+    row = next(row for row in snapshot.rows if row.key == LEARNING_OFF_KEY)
+    assert row.value == label
+    assert row.available == (saved_raw != b"bad")
+    assert row.choices == (() if saved_raw == b"bad" else ("Off", "On"))
+    for tick in range(2):
+      now = 1_000_000_000 + tick * 250_000_000
+      feed(controls, now, 0)
+      controls.sm.update_msgs(now / 1e9, [learned_event(cp, now)])
+    assert controls.sm.all_checks(["lateralTorqueParameters"])
+    with patch.object(controls.LaC, "update_torque_parameters") as update:
+      command, _ = controls.state_control()
+      assert command.latActive
+      assert update.call_count == int(allowed)
+    assert raw_saved(saved, LEARNING_OFF_KEY) == saved_raw
+
+
+@pytest.mark.parametrize("initial_mode", [ControllerMode.STANDARD, ControllerMode.STARPILOT])
+def test_actual_ui_controller_switch_preserves_learning_and_explicit_toggle(initial_mode):
+  with OpenpilotPrefix():
+    cp = factory()
+    saved = Params()
+    saved.put(DOCUMENT_KEY, json.loads(replace_mode(None, cp, initial_mode)), block=True)
+    owner = FeatureSettingsOwner(saved, lambda group: group == "preferences",
+                                 vehicle_fingerprint=lambda: str(cp.carFingerprint), vehicle_params=lambda: cp)
+
+    def row(key):
+      state = owner.snapshot("torque", parked=False, system_long=False, lateral_context=False, metric=False)
+      return next(row for row in state.rows if row.key == key)
+
+    controller_row = row(DOCUMENT_KEY)
+    request = row_change(controller_row)
+    assert request is not None
+    assert owner.apply(request)
+    # RAV4's native-learning default remains enabled across both controller choices.
+    assert raw_saved(saved, LEARNING_OFF_KEY) is None
+    learning_row = row(LEARNING_OFF_KEY)
+    assert learning_row.available and learning_row.value == "On"
+    request = row_change(learning_row)
+    assert request is not None
+    assert owner.apply(request)
+    assert raw_saved(saved, LEARNING_OFF_KEY) == b"1"
+    learning_row = row(LEARNING_OFF_KEY)
+    assert learning_row.value == "Off" and learning_row.available
+    request = row_change(learning_row)
+    assert request is not None
+    assert owner.apply(request)
+    assert raw_saved(saved, LEARNING_OFF_KEY) == b"0"

openpilot/starpilot/lateral/tests/test_torque_learning.py

@@ -66,7 +66,7 @@ class TestTorqueLearning(unittest.TestCase):
       params.put_bool(LEARNING_OFF_KEY, True, block=True)
       for vehicle in (HYUNDAI.KIA_EV6, HYUNDAI.HYUNDAI_SONATA):
         self.assertFalse(learning_allowed(params, cp_for(vehicle)))
-      self.assertTrue(learning_allowed(params, cp_for(TOYOTA.TOYOTA_RAV4_TSS2)))
+      self.assertFalse(learning_allowed(params, cp_for(TOYOTA.TOYOTA_RAV4_TSS2)))
 
   def test_learner_does_not_restore_collect_estimate_or_overwrite_disabled_cache(self):
     for vehicle in VEHICLES:

openpilot/starpilot/lateral/torque_extension.py

@@ -43,6 +43,10 @@ def create_extension(parent, cp, mode, selected, *, turn_assist=False):
     from openpilot.starpilot.lateral.genesis_g70_policy import GenesisG70TorquePolicy, supported_cp
     if supported_cp(cp):
       return TorqueExtension(GenesisG70TorquePolicy(parent, cp))
+  if selected == 'rav4_tss2':
+    from openpilot.starpilot.lateral.rav4_tss2_policy import Rav4TSS2TorquePolicy, supported_cp
+    if supported_cp(cp):
+      return TorqueExtension(Rav4TSS2TorquePolicy(parent, cp))
   if selected == 'corolla_tss2':
     from openpilot.starpilot.lateral.corolla_tss2_policy import CorollaTSS2TorquePolicy, supported_cp
     if supported_cp(cp):

openpilot/starpilot/ui/controller_feature.py

@@ -7,6 +7,7 @@ import json
 
 from openpilot.starpilot.lateral.controller_selection import (
   DOCUMENT_KEY, LEARNING_OFF_KEY, MAX_DOCUMENT_BYTES, ControllerMode, policy_for, replace_mode, selection_from_bytes,
+  controller_uses_native_learning,
 )
 from openpilot.starpilot.saved_document import commit_exact
 from openpilot.starpilot.saved_source import read_saved
@@ -149,12 +150,12 @@ class ControllerFeature:
     selection = selection_from_bytes(CP, controller)
     raw, readable = read_saved(self.params, LEARNING_OFF_KEY, 1)
     valid = readable and raw in (None, b'0', b'1') and controller_readable and selection.source != 'invalid'
-    starpilot = selection.mode == ControllerMode.STARPILOT
-    value = ('Off' if starpilot or raw == b'1' else 'On') if valid else 'Invalid saved preference'
+    fixed_vehicle_tune = not controller_uses_native_learning(selection)
+    value = ('Off' if fixed_vehicle_tune or raw == b'1' else 'On') if valid else 'Invalid saved preference'
     reason = ('StarPilot uses the vehicle tune. ' + TUNING_GUIDANCE
-              if starpilot else 'Saved for the next drive. Turn on only if you want Stock Controller to learn torque values.')
+              if fixed_vehicle_tune else 'Saved for the next drive. Turn on if you want this controller to learn torque values.')
     return FeatureRow(LEARNING_OFF_KEY, 'Automatic Steering Learning', value, raw, ('Off', 'On') if valid else (),
-                      available=valid and not starpilot and self.authority('preferences'),
+                      available=valid and not fixed_vehicle_tune and self.authority('preferences'),
                       reason=reason if valid else 'Saved controller or learning preference is unreadable or invalid.',
                       capability=capability, dependencies=((DOCUMENT_KEY, controller),), default_value="On")
 
@@ -202,7 +203,8 @@ class ControllerFeature:
       return (self.authority('preferences') and self.vehicle_fingerprint() == request.vehicle_fingerprint and
               self.capability() == request.capability and readable and current == learning)
 
-    if learning != b'1' and (MODES[request.value] == ControllerMode.STARPILOT or selected.mode == ControllerMode.STARPILOT):
+    if learning != b'1' and (not controller_uses_native_learning(replace(selected, mode=MODES[request.value])) or
+                             not controller_uses_native_learning(selected)):
       def authorize_learning_off() -> bool:
         current, readable = read_saved(self.params, DOCUMENT_KEY, MAX_DOCUMENT_BYTES)
         return authorized() and readable and current == request.expected