Traffic Mode v2

This commit is contained in:
whoisdomi
2026-07-17 10:41:08 -05:00
committed by firestar5683
parent 8d7ba284d7
commit b70d0941c9
13 changed files with 277 additions and 21 deletions
+5 -5
View File
@@ -622,12 +622,12 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
{"TinygradUpdateAvailable", {PERSISTENT, BOOL, "0", "0", 1}},
{"ToyotaDoors", {PERSISTENT, BOOL, "1", "0", 0}},
{"TrailerLoad", {PERSISTENT, INT, "0", "0", 2}},
{"TrafficFollow", {PERSISTENT, FLOAT, "0.5", "0.5", 2}},
{"TrafficJerkAcceleration", {PERSISTENT, FLOAT, "50.0", "50.0", 3}},
{"TrafficFollow", {PERSISTENT, FLOAT, "0.75", "0.75", 2}},
{"TrafficJerkAcceleration", {PERSISTENT, FLOAT, "100.0", "100.0", 3}},
{"TrafficJerkDanger", {PERSISTENT, FLOAT, "100.0", "100.0", 3}},
{"TrafficJerkDeceleration", {PERSISTENT, FLOAT, "50.0", "50.0", 3}},
{"TrafficJerkSpeed", {PERSISTENT, FLOAT, "50.0", "50.0", 3}},
{"TrafficJerkSpeedDecrease", {PERSISTENT, FLOAT, "50.0", "50.0", 3}},
{"TrafficJerkDeceleration", {PERSISTENT, FLOAT, "100.0", "100.0", 3}},
{"TrafficJerkSpeed", {PERSISTENT, FLOAT, "100.0", "100.0", 3}},
{"TrafficJerkSpeedDecrease", {PERSISTENT, FLOAT, "100.0", "100.0", 3}},
{"TruckTuning", {PERSISTENT, BOOL, "0", "0", 3}},
{"TuningLevel", {PERSISTENT, INT, "0", "0", 0}},
{"TuningLevelConfirmed", {PERSISTENT, BOOL, "0", "0", 0}},
+2 -1
View File
@@ -423,7 +423,8 @@ class Controls:
pid_accel_limits = self.CI.get_pid_accel_limits(self.CP, CS.vEgo, CS.vCruise * CV.KPH_TO_MS)
self.LoC.experimental_mode = bool(self.sm['selfdriveState'].experimentalMode)
actuators.accel = float(min(self.LoC.update(CC.longActive, CS, long_plan.aTarget, long_plan.shouldStop, pid_accel_limits,
self.starpilot_toggles, has_lead=long_plan.hasLead),
self.starpilot_toggles, has_lead=long_plan.hasLead,
traffic_mode_enabled=self.sm['starpilotCarState'].trafficModeEnabled),
self.starpilot_toggles.max_desired_acceleration))
# Steering PID loop and lateral MPC
+6 -2
View File
@@ -227,7 +227,7 @@ class LongControl:
positive_cap = interp(a_target, [-1.5, -0.6, -0.1], [0.0, 0.0, 0.05])
return min(output_accel, float(positive_cap))
def update(self, active, CS, a_target, should_stop, accel_limits, starpilot_toggles, has_lead=False):
def update(self, active, CS, a_target, should_stop, accel_limits, starpilot_toggles, has_lead=False, traffic_mode_enabled=False):
"""Update longitudinal control. This updates the state machine and runs a PID loop"""
self.pid.neg_limit = accel_limits[0]
self.pid.pos_limit = accel_limits[1]
@@ -250,7 +250,11 @@ class LongControl:
self.reset(preserve_stop_release=True)
elif self.long_control_state == LongCtrlState.starting:
if getattr(starpilot_toggles, "custom_accel_profile", False):
if traffic_mode_enabled:
# Traffic Mode has its own soft launch curve (a_target); bypass the raw
# StartAccel kick used elsewhere so launches stay within the traffic cap.
output_accel = clip(a_target, 0.0, starpilot_toggles.startAccel)
elif getattr(starpilot_toggles, "custom_accel_profile", False):
output_accel = clip(a_target, 0.0, starpilot_toggles.startAccel)
else:
output_accel = starpilot_toggles.startAccel
@@ -170,6 +170,58 @@ def test_starting_accel_obeys_a_target_cap_when_custom_profile_enabled():
assert output_accel == 0.1
def test_starting_accel_obeys_a_target_cap_when_traffic_mode_enabled():
CP = car.CarParams.new_message(startingState=True, vEgoStarting=0.5)
CP.longitudinalTuning.kpBP = [0.0]
CP.longitudinalTuning.kpV = [0.1]
CP.longitudinalTuning.kiBP = [0.0]
CP.longitudinalTuning.kiV = [0.03]
lc = LongControl(CP)
CS = car.CarState.new_message(vEgo=0.0, aEgo=0.0, brakePressed=False)
CS.cruiseState.standstill = False
# Large manually-tuned startAccel override (e.g. 3.5) should not fire a raw
# launch kick while Traffic Mode is active; output must track the soft a_target.
output_accel = lc.update(
active=True,
CS=CS,
a_target=1.10,
should_stop=False,
accel_limits=(-3.0, 4.0),
starpilot_toggles=make_toggles(startAccel=3.5),
traffic_mode_enabled=True,
)
assert lc.long_control_state == LongCtrlState.starting
assert output_accel == pytest.approx(1.10)
def test_starting_accel_uses_raw_start_accel_when_traffic_mode_disabled():
CP = car.CarParams.new_message(startingState=True, vEgoStarting=0.5)
CP.longitudinalTuning.kpBP = [0.0]
CP.longitudinalTuning.kpV = [0.1]
CP.longitudinalTuning.kiBP = [0.0]
CP.longitudinalTuning.kiV = [0.03]
lc = LongControl(CP)
CS = car.CarState.new_message(vEgo=0.0, aEgo=0.0, brakePressed=False)
CS.cruiseState.standstill = False
output_accel = lc.update(
active=True,
CS=CS,
a_target=1.10,
should_stop=False,
accel_limits=(-3.0, 4.0),
starpilot_toggles=make_toggles(startAccel=3.5),
traffic_mode_enabled=False,
)
assert lc.long_control_state == LongCtrlState.starting
assert output_accel == pytest.approx(3.5)
def test_update_requires_sustained_moderate_positive_target_to_leave_stopping():
CP = car.CarParams.new_message(startingState=True, vEgoStarting=0.5)
CP.longitudinalTuning.kpBP = [0.0]
@@ -4,11 +4,14 @@ import pytest
from openpilot.common.constants import CV
from openpilot.selfdrive.controls.lib.longitudinal_planner import A_CRUISE_MIN
from openpilot.starpilot.common.accel_profile import ACCELERATION_PROFILES, DECELERATION_PROFILES
from openpilot.starpilot.common.accel_profile import A_CRUISE_MAX_BP_CUSTOM, ACCELERATION_PROFILES, DECELERATION_PROFILES
from openpilot.starpilot.controls.lib.starpilot_acceleration import (
A_CRUISE_MIN_ECO,
A_CRUISE_MIN_TRAFFIC,
StarPilotAcceleration,
get_max_accel_eco,
get_max_accel_standard,
get_max_accel_traffic,
get_slc_shaped_min_accel,
)
@@ -163,3 +166,52 @@ def test_truck_tuning_standard_profile_uses_proven_cruise_limits():
def test_truck_tuning_standard_profile_limits_highway_run_up():
assert get_max_accel_standard(40.0, ev_tuning=False, truck_tuning=True) == pytest.approx(0.35)
def test_traffic_curve_anchors():
assert get_max_accel_traffic(0.0) == pytest.approx(1.10)
assert get_max_accel_traffic(10.0) == pytest.approx(0.67)
assert get_max_accel_traffic(25.0) == pytest.approx(0.34)
assert get_max_accel_traffic(40.0) == pytest.approx(0.23)
def test_traffic_curve_softer_than_eco():
for v in A_CRUISE_MAX_BP_CUSTOM:
assert get_max_accel_traffic(v) < get_max_accel_eco(v, ev_tuning=True)
assert get_max_accel_traffic(v) < get_max_accel_eco(v, ev_tuning=False)
def test_traffic_mode_overrides_custom_accel_profile():
accel = StarPilotAcceleration(FakePlanner(v_cruise=25.0))
sm = make_sm(traffic_mode=True)
accel.update(5.0, sm, make_toggles(custom_accel_profile=True, custom_accel_profile_values=[6.0] * 7))
assert accel.max_accel == pytest.approx(get_max_accel_traffic(5.0))
def test_traffic_mode_skips_human_acceleration_shaping():
accel = StarPilotAcceleration(FakePlanner(v_cruise=2.0))
sm = make_sm(traffic_mode=True)
accel.update(1.0, sm, make_toggles(human_acceleration=True))
assert accel.max_accel == pytest.approx(get_max_accel_traffic(1.0))
def test_traffic_mode_sets_soft_cruise_decel_floor():
accel = StarPilotAcceleration(FakePlanner(v_cruise=25.0))
sm = make_sm(traffic_mode=True)
accel.update(5.0, sm, make_toggles())
assert accel.min_accel == pytest.approx(A_CRUISE_MIN_TRAFFIC)
def test_force_coast_wins_over_traffic_mode_decel():
accel = StarPilotAcceleration(FakePlanner(v_cruise=25.0))
sm = make_sm(traffic_mode=True, force_coast=True)
accel.update(5.0, sm, make_toggles())
assert accel.min_accel == pytest.approx(A_CRUISE_MIN_ECO)
@@ -1106,7 +1106,7 @@ class StarPilotLongitudinalLayout(_SettingsPage):
self._navigate_to(panel_name)
def _build_personality_profile_rows(self, profile: str) -> list[SettingRow]:
follow_min = 1.0 if profile == "Traffic" else 0.5
follow_min = 0.5
follow_max = 2.5 if profile == "Traffic" else 3.0
p = profile
rows = [
+4
View File
@@ -45,6 +45,10 @@ A_CRUISE_MAX_VALS_STANDARD_GAS = [2.00, 1.80, 1.55, 1.30, 1.05, 0.85, 0.55]
A_CRUISE_MAX_VALS_SPORT_GAS = [2.50, 2.25, 1.95, 1.60, 1.30, 1.05, 0.75]
A_CRUISE_MAX_VALS_SPORT_PLUS_GAS = [3.50, 3.20, 2.80, 2.35, 1.90, 1.55, 1.15]
# Traffic Mode: bumper-to-bumper stop-and-go, softer than Eco at every breakpoint.
# Single curve for all vehicle types, derived from manual stop-and-go driving logs.
A_CRUISE_MAX_VALS_TRAFFIC_ALL = [1.10, 0.87, 0.67, 0.53, 0.44, 0.34, 0.23]
A_CRUISE_MAX_VALS_ECO_TRUCK = [3.00, 1.05, 0.60, 0.50, 0.50, 0.45, 0.35]
A_CRUISE_MAX_VALS_STANDARD_TRUCK = [6.00, 1.10, 0.70, 0.60, 0.55, 0.45, 0.35]
A_CRUISE_MAX_VALS_SPORT_TRUCK = [6.00, 1.15, 0.75, 0.70, 0.60, 0.50, 0.40]
+6 -6
View File
@@ -856,12 +856,12 @@ class StarPilotVariables:
relaxed_follow_low = float(self.get_value("RelaxedFollow", cast=float, condition=toggle.custom_personalities, min=1, max=MAX_T_FOLLOW))
relaxed_follow_high = float(self.get_value("RelaxedFollowHigh", cast=float, condition=toggle.custom_personalities, min=1, max=MAX_T_FOLLOW))
toggle.relaxed_follow = [relaxed_follow_low, relaxed_follow_high]
toggle.traffic_mode_jerk_acceleration = [self.get_value("TrafficJerkAcceleration", cast=float, condition=toggle.custom_personalities, conversion=0.01, min=0.25, max=2.0), toggle.aggressive_jerk_acceleration]
toggle.traffic_mode_jerk_deceleration = [self.get_value("TrafficJerkDeceleration", cast=float, condition=toggle.custom_personalities, conversion=0.01, min=0.25, max=2.0), toggle.aggressive_jerk_deceleration]
toggle.traffic_mode_jerk_danger = [self.get_value("TrafficJerkDanger", cast=float, condition=toggle.custom_personalities, conversion=0.01, min=0.25, max=2.0), toggle.aggressive_jerk_danger]
toggle.traffic_mode_jerk_speed = [self.get_value("TrafficJerkSpeed", cast=float, condition=toggle.custom_personalities, conversion=0.01, min=0.25, max=2.0), toggle.aggressive_jerk_speed]
toggle.traffic_mode_jerk_speed_decrease = [self.get_value("TrafficJerkSpeedDecrease", cast=float, condition=toggle.custom_personalities, conversion=0.01, min=0.25, max=2.0), toggle.aggressive_jerk_speed_decrease]
toggle.traffic_mode_follow = [float(self.get_value("TrafficFollow", cast=float, condition=toggle.custom_personalities, min=0.5, max=MAX_T_FOLLOW)), toggle.aggressive_follow[0]]
toggle.traffic_mode_jerk_acceleration = [self.get_value("TrafficJerkAcceleration", cast=float, condition=toggle.custom_personalities, conversion=0.01, min=0.25, max=2.0), toggle.relaxed_jerk_acceleration]
toggle.traffic_mode_jerk_deceleration = [self.get_value("TrafficJerkDeceleration", cast=float, condition=toggle.custom_personalities, conversion=0.01, min=0.25, max=2.0), toggle.relaxed_jerk_deceleration]
toggle.traffic_mode_jerk_danger = [self.get_value("TrafficJerkDanger", cast=float, condition=toggle.custom_personalities, conversion=0.01, min=0.25, max=2.0), toggle.relaxed_jerk_danger]
toggle.traffic_mode_jerk_speed = [self.get_value("TrafficJerkSpeed", cast=float, condition=toggle.custom_personalities, conversion=0.01, min=0.25, max=2.0), toggle.relaxed_jerk_speed]
toggle.traffic_mode_jerk_speed_decrease = [self.get_value("TrafficJerkSpeedDecrease", cast=float, condition=toggle.custom_personalities, conversion=0.01, min=0.25, max=2.0), toggle.relaxed_jerk_speed_decrease]
toggle.traffic_mode_follow = [float(self.get_value("TrafficFollow", cast=float, condition=toggle.custom_personalities, min=0.5, max=MAX_T_FOLLOW)), toggle.relaxed_follow[0]]
custom_themes = self.get_value("CustomThemes")
toggle.boot_logo = self.get_value("BootLogo", cast=None, default="starpilot")
@@ -8,6 +8,7 @@ from openpilot.selfdrive.controls.lib.longitudinal_planner import A_CRUISE_MIN,
from openpilot.starpilot.common.accel_profile import (
ACCELERATION_PROFILES,
A_CRUISE_MAX_VALS_TRAFFIC_ALL,
DECELERATION_PROFILES,
coerce_custom_accel_profile_values,
get_accel_profile_curve_values,
@@ -56,6 +57,7 @@ def akima_interp(x, xp, fp):
A_CRUISE_MIN_ECO = A_CRUISE_MIN / 2
A_CRUISE_MIN_SPORT = A_CRUISE_MIN * 2
A_CRUISE_MIN_TRAFFIC = A_CRUISE_MIN * 0.35 # cruise-decel floor only; MPC lead braking keeps full ACCEL_MIN authority
SLC_COAST_WINDOW_BP = [0.0, 10.0, 20.0, 35.0]
SLC_COAST_WINDOW_BASE = [0.20, 0.40, 0.65, 1.10]
SLC_EXCESS_SCALE_BP = [0.0, 10.0, 20.0, 35.0]
@@ -84,6 +86,9 @@ def get_max_accel_sport(v_ego, ev_tuning=True, truck_tuning=False):
def get_max_accel_standard(v_ego, ev_tuning=True, truck_tuning=False):
return interpolate_accel_profile(v_ego, get_accel_profile_curve_values(ACCELERATION_PROFILES["STANDARD"], ev_tuning, truck_tuning))
def get_max_accel_traffic(v_ego):
return interpolate_accel_profile(v_ego, A_CRUISE_MAX_VALS_TRAFFIC_ALL)
def get_max_accel_custom(v_ego, custom_curve, acceleration_profile, ev_tuning=True, truck_tuning=False):
curve_values = coerce_custom_accel_profile_values(custom_curve, acceleration_profile, ev_tuning, truck_tuning)
return interpolate_accel_profile(v_ego, curve_values)
@@ -147,10 +152,10 @@ class StarPilotAcceleration:
getattr(starpilot_toggles, "deceleration_profile", DECELERATION_PROFILES["STANDARD"])
)
if custom_accel_profile:
if sm["starpilotCarState"].trafficModeEnabled:
self.max_accel = get_max_accel_traffic(v_ego)
elif custom_accel_profile:
self.max_accel = get_max_accel_custom(v_ego, custom_accel_profile_values, starpilot_toggles.acceleration_profile, ev_tuning, truck_tuning)
elif sm["starpilotCarState"].trafficModeEnabled:
self.max_accel = get_max_accel_standard(v_ego, ev_tuning, truck_tuning)
elif starpilot_toggles.map_acceleration and (eco_gear or sport_gear):
if eco_gear:
self.max_accel = get_max_accel_eco(v_ego, ev_tuning, truck_tuning)
@@ -171,6 +176,8 @@ class StarPilotAcceleration:
if sm["starpilotCarState"].forceCoast:
self.min_accel = A_CRUISE_MIN_ECO
elif sm["starpilotCarState"].trafficModeEnabled:
self.min_accel = A_CRUISE_MIN_TRAFFIC
elif starpilot_toggles.map_deceleration and (eco_gear or sport_gear):
if eco_gear:
self.min_accel = A_CRUISE_MIN_ECO
@@ -752,7 +752,7 @@
{
"key": "TrafficFollow",
"label": "Following Distance",
"description": "The minimum following distance to the lead vehicle in \"Traffic Mode\". openpilot blends between this value and the \"Aggressive\" profile as speed increases. Increase for more space; decrease for tighter gaps.",
"description": "The minimum following distance to the lead vehicle in \"Traffic Mode\". openpilot blends between this value and the \"Relaxed\" profile as speed increases. Increase for more space; decrease for tighter gaps.",
"data_type": "float",
"ui_type": "numeric",
"min": 0.5,
@@ -116,7 +116,7 @@ StarPilotLongitudinalPanel::StarPilotLongitudinalPanel(StarPilotSettingsWindow *
{"CustomPersonalities", tr("Driving Personalities"), tr("<b>Customize the \"Driving Personalities\"</b> to better match your driving style."), "../../starpilot/assets/toggle_icons/icon_personality.png"},
{"TrafficPersonalityProfile", tr("Traffic Mode"), tr("<b>Customize the \"Traffic Mode\" personality profile.</b> Designed for stop-and-go driving."), "../../starpilot/assets/stock_theme/distance_icons/traffic.png"},
{"TrafficFollow", tr("Following Distance"), tr("<b>The minimum following distance to the lead vehicle in \"Traffic Mode\".</b> openpilot blends between this value and the \"Aggressive\" profile as speed increases. Increase for more space; decrease for tighter gaps."), ""},
{"TrafficFollow", tr("Following Distance"), tr("<b>The minimum following distance to the lead vehicle in \"Traffic Mode\".</b> openpilot blends between this value and the \"Relaxed\" profile as speed increases. Increase for more space; decrease for tighter gaps."), ""},
{"TrafficJerkAcceleration", tr("Acceleration Smoothness"), tr("<b>How smoothly openpilot accelerates in \"Traffic Mode\".</b> Increase for gentler starts; decrease for faster but more abrupt takeoffs."), ""},
{"TrafficJerkDeceleration", tr("Braking Smoothness"), tr("<b>How smoothly openpilot brakes in \"Traffic Mode\".</b> Increase for gentler stops; decrease for quicker but sharper braking."), ""},
{"TrafficJerkDanger", tr("Safety Gap Bias"), tr("<b>How much extra space openpilot keeps from the vehicle ahead in \"Traffic Mode\".</b> Increase for larger gaps and more cautious following; decrease for tighter gaps and closer following."), ""},
+65
View File
@@ -61,6 +61,8 @@ LEGACY_BOLT_FP_MIGRATION_FLAG = Path("/data") / "legacy_bolt_fp_migration_v1"
STARPILOT_DEFAULTS_PARITY_MIGRATION_FLAG = Path("/data") / "starpilot_defaults_parity_v1"
STARPILOT_HUMANLIKE_DISABLE_MIGRATION_FLAG = Path("/data") / "starpilot_humanlike_disable_v1"
STARPILOT_CLUSTER_OFFSET_MIGRATION_FLAG = Path("/data") / "starpilot_cluster_offset_v1"
STARPILOT_TRAFFIC_SMOOTH_MIGRATION_FLAG = Path("/data") / "starpilot_traffic_smooth_v1"
STARPILOT_TRAFFIC_FOLLOW_MIGRATION_FLAG = Path("/data") / "starpilot_traffic_follow_v1"
STARPILOT_PARAM_RENAME_MIGRATION_FLAG = Path("/data") / "starpilot_param_rename_v1"
STARPILOT_PARAM_CANONICALIZATION_MIGRATION_FLAG = Path("/data") / "starpilot_param_canonicalization_v1"
STARPILOT_PC_ROOT_MIGRATION_FLAG = Path("/data") / "starpilot_pc_root_v1"
@@ -603,6 +605,67 @@ def migrate_cluster_offset_default(params: Params, params_cache: Params) -> None
cloudlog.exception(f"Failed to write migration flag: {STARPILOT_CLUSTER_OFFSET_MIGRATION_FLAG}")
def migrate_traffic_mode_smooth_defaults(params: Params, params_cache: Params) -> None:
# Traffic Mode was repurposed from an aggressive city mode (jerk 50%) into a smooth
# bumper-to-bumper mode with Relaxed-parity jerk defaults (100%). Rewrite persisted
# legacy defaults only; user-tuned values are preserved.
if STARPILOT_TRAFFIC_SMOOTH_MIGRATION_FLAG.exists():
return
migrated_keys: list[str] = []
for key in ("TrafficJerkAcceleration", "TrafficJerkDeceleration", "TrafficJerkSpeed", "TrafficJerkSpeedDecrease"):
# Decide off the real params store only: the boot sync mirrors real -> cache, so a
# cache-first check could clobber a user override with a stale cached default.
raw_value = _read_raw_param_bytes(params, key)
if not raw_value:
continue
try:
parsed_value = float(raw_value.decode("utf-8", errors="strict").strip())
except Exception:
continue
if abs(parsed_value - 50.0) < 1e-6:
params.put_float(key, 100.0)
params_cache.put_float(key, 100.0)
migrated_keys.append(key)
if migrated_keys:
cloudlog.warning(f"Applied one-time Traffic Mode smooth-defaults migration for {migrated_keys}")
try:
STARPILOT_TRAFFIC_SMOOTH_MIGRATION_FLAG.parent.mkdir(parents=True, exist_ok=True)
STARPILOT_TRAFFIC_SMOOTH_MIGRATION_FLAG.write_text(f"{datetime.datetime.now(datetime.UTC).isoformat()}\n")
except Exception:
cloudlog.exception(f"Failed to write migration flag: {STARPILOT_TRAFFIC_SMOOTH_MIGRATION_FLAG}")
def migrate_traffic_follow_default(params: Params, params_cache: Params) -> None:
# TrafficFollow's initial smooth-mode default (0.5s) proved too tight in on-road
# testing (frequent closing-on-lead); raised to 0.75s. Rewrite persisted legacy
# 0.5 only; user-tuned values are preserved.
if STARPILOT_TRAFFIC_FOLLOW_MIGRATION_FLAG.exists():
return
raw_value = _read_raw_param_bytes(params, "TrafficFollow")
if raw_value:
try:
parsed_value = float(raw_value.decode("utf-8", errors="strict").strip())
except Exception:
parsed_value = None
if parsed_value is not None and abs(parsed_value - 0.5) < 1e-6:
params.put_float("TrafficFollow", 0.75)
params_cache.put_float("TrafficFollow", 0.75)
cloudlog.warning("Applied one-time TrafficFollow migration from 0.5 to 0.75")
try:
STARPILOT_TRAFFIC_FOLLOW_MIGRATION_FLAG.parent.mkdir(parents=True, exist_ok=True)
STARPILOT_TRAFFIC_FOLLOW_MIGRATION_FLAG.write_text(f"{datetime.datetime.now(datetime.UTC).isoformat()}\n")
except Exception:
cloudlog.exception(f"Failed to write migration flag: {STARPILOT_TRAFFIC_FOLLOW_MIGRATION_FLAG}")
def _read_raw_param_bytes(params: Params, key: str | bytes):
try:
path = params.get_param_path(key)
@@ -822,6 +885,8 @@ def manager_init() -> None:
migrate_starpilot_default_parity(params, params_cache)
migrate_disable_humanlike_defaults(params, params_cache)
migrate_cluster_offset_default(params, params_cache)
migrate_traffic_mode_smooth_defaults(params, params_cache)
migrate_traffic_follow_default(params, params_cache)
last_timing = _log_boot_timing("manager_init", "starpilot_migrations", manager_init_start, last_timing)
# set unset params to their default value
+71
View File
@@ -345,6 +345,77 @@ class TestManager:
assert params.get("ClusterOffset") == "1.02"
assert params_cache.get("ClusterOffset") is None
def test_migrate_traffic_mode_smooth_defaults_resets_legacy_default_only(self, tmp_path, monkeypatch):
monkeypatch.setattr(manager, "STARPILOT_TRAFFIC_SMOOTH_MIGRATION_FLAG", tmp_path / "starpilot_traffic_smooth_v1")
params = FileBackedFakeParams(tmp_path / "params", {
"TrafficJerkAcceleration": 50.0,
"TrafficJerkDeceleration": 50.0,
"TrafficJerkSpeed": 50.0,
})
params_cache = FileBackedFakeParams(tmp_path / "cache", {})
manager.migrate_traffic_mode_smooth_defaults(params, params_cache)
for key in ("TrafficJerkAcceleration", "TrafficJerkDeceleration", "TrafficJerkSpeed"):
assert params.get(key) == "100.0"
assert params_cache.get(key) == "100.0"
# unset keys stay unset so the new compiled default applies on its own
assert params.get("TrafficJerkSpeedDecrease") is None
assert manager.STARPILOT_TRAFFIC_SMOOTH_MIGRATION_FLAG.exists()
def test_migrate_traffic_mode_smooth_defaults_preserves_custom_values(self, tmp_path, monkeypatch):
monkeypatch.setattr(manager, "STARPILOT_TRAFFIC_SMOOTH_MIGRATION_FLAG", tmp_path / "starpilot_traffic_smooth_v1")
params = FileBackedFakeParams(tmp_path / "params", {
"TrafficJerkAcceleration": 80.0,
})
params_cache = FileBackedFakeParams(tmp_path / "cache", {})
manager.migrate_traffic_mode_smooth_defaults(params, params_cache)
assert params.get("TrafficJerkAcceleration") == "80.0"
assert params_cache.get("TrafficJerkAcceleration") is None
def test_migrate_traffic_follow_default_resets_legacy_default_only(self, tmp_path, monkeypatch):
monkeypatch.setattr(manager, "STARPILOT_TRAFFIC_FOLLOW_MIGRATION_FLAG", tmp_path / "starpilot_traffic_follow_v1")
params = FileBackedFakeParams(tmp_path / "params", {
"TrafficFollow": 0.5,
})
params_cache = FileBackedFakeParams(tmp_path / "cache", {})
manager.migrate_traffic_follow_default(params, params_cache)
assert params.get("TrafficFollow") == "0.75"
assert params_cache.get("TrafficFollow") == "0.75"
assert manager.STARPILOT_TRAFFIC_FOLLOW_MIGRATION_FLAG.exists()
def test_migrate_traffic_follow_default_preserves_custom_values(self, tmp_path, monkeypatch):
monkeypatch.setattr(manager, "STARPILOT_TRAFFIC_FOLLOW_MIGRATION_FLAG", tmp_path / "starpilot_traffic_follow_v1")
params = FileBackedFakeParams(tmp_path / "params", {
"TrafficFollow": 1.2,
})
params_cache = FileBackedFakeParams(tmp_path / "cache", {})
manager.migrate_traffic_follow_default(params, params_cache)
assert params.get("TrafficFollow") == "1.2"
assert params_cache.get("TrafficFollow") is None
def test_migrate_traffic_mode_smooth_defaults_runs_once(self, tmp_path, monkeypatch):
monkeypatch.setattr(manager, "STARPILOT_TRAFFIC_SMOOTH_MIGRATION_FLAG", tmp_path / "starpilot_traffic_smooth_v1")
params = FileBackedFakeParams(tmp_path / "params", {"TrafficJerkAcceleration": 50.0})
params_cache = FileBackedFakeParams(tmp_path / "cache", {})
manager.migrate_traffic_mode_smooth_defaults(params, params_cache)
params.put_float("TrafficJerkAcceleration", 50.0)
manager.migrate_traffic_mode_smooth_defaults(params, params_cache)
assert params.get("TrafficJerkAcceleration") == "50.0"
def test_cleanup_inaccessible_msgq_files_removes_only_blocked_files(self, tmp_path, monkeypatch):
healthy = tmp_path / "msgq_deviceState"
blocked = tmp_path / "msgq_gpsLocation"