StarPort: Phone ping as an experimental on-off toggle

This commit is contained in:
Danny
2026-11-08 16:04:59 -08:00
parent 3b1c908070
commit c37cee5e80
6 changed files with 73 additions and 12 deletions
@@ -73,12 +73,11 @@ class DeveloperSettingsOwner:
rows.append(FeatureRow(STARPORT_EXPERIMENTS_KEY, "Try experimental fixes", "On" if experiments else "Off",
b"1" if experiments else b"0", ("Off", "On"), available=parked,
reason=STARPORT_EXPERIMENTS_DESCRIPTION if parked else held, default_value="Off"))
for rule in head_unit_fixes.RULES:
if rule.status == head_unit_fixes.EXPERIMENTAL:
on = rule.id in chosen
rows.append(FeatureRow(STARPORT_EXPERIMENT_PREFIX + rule.id, rule.label, "On" if on else "Off", b"1" if on else b"0",
("Off", "On"), available=parked, default_value="Off",
reason=f"Experimental. {rule.evidence.failure} Applies from the next StarPort connection." if parked else held))
for entry in head_unit_fixes.experimental_entries():
on = entry["id"] in chosen
rows.append(FeatureRow(STARPORT_EXPERIMENT_PREFIX + entry["id"], entry["label"], "On" if on else "Off", b"1" if on else b"0",
("Off", "On"), available=parked, default_value="Off",
reason=f"Experimental. {entry['evidence']} Applies from the next StarPort connection." if parked else held))
return tuple(rows)
def _driver_monitoring_reset_allowed(self):
@@ -172,7 +171,7 @@ class DeveloperSettingsOwner:
def _apply_experiment_rule(self, request):
rule_id = request.key.removeprefix(STARPORT_EXPERIMENT_PREFIX)
status, client = self._starport_status(), self.starport
if status is None or client is None or rule_id not in head_unit_fixes.rule_ids() or not self.authorized():
if status is None or client is None or rule_id not in head_unit_fixes.experiment_ids() or not self.authorized():
return False
on = rule_id in (status.get("head_unit_experiment_rules") or ())
if request.value not in ("Off", "On") or request.expected != (b"1" if on else b"0"):
@@ -273,6 +273,11 @@ class DeveloperSettingsTest(unittest.TestCase):
self.assertFalse(owner.apply(FeatureSettingsRequest(rule_key, b"0", "On"))) # stale view
self.assertTrue(owner.apply(FeatureSettingsRequest(rule_key, b"1", "Off")))
self.assertEqual(service.rules, [])
mavlou = STARPORT_EXPERIMENT_PREFIX + "phone_ping"
self.assertEqual(rows[keys.index(mavlou)].value, "Off")
self.assertTrue(owner.apply(FeatureSettingsRequest(mavlou, b"0", "On")))
self.assertEqual(service.rules, ["phone_ping"])
self.assertTrue(owner.apply(FeatureSettingsRequest(mavlou, b"1", "Off")))
self.assertFalse(owner.apply(FeatureSettingsRequest(STARPORT_EXPERIMENT_PREFIX + "unknown", b"0", "On")))
self.parked = False
self.assertFalse(owner.apply(FeatureSettingsRequest(rule_key, b"0", "On")))
@@ -81,7 +81,8 @@ EXPERIMENTS = {
"start_response_first": ("Answer start request first", {"start_response_first"}),
"version_endpoint": ("Use version endpoint", {"version_endpoint"}),
"phone_ping": ("Ping like a phone", {"phone_ping"}),
"all": ("All of these", {"start_response_first", "version_endpoint", "phone_ping"}),
"all": ("All of these", {"start_response_first", "version_endpoint"}),
"all_with_ping": ("All of these and phone ping", {"start_response_first", "version_endpoint", "phone_ping"}),
}
# Frames that carry credentials or the vehicle identifier are never dumped as hex.
HEX_LOG_SKIP = (WIFI_INFO_RESPONSE, WIFI_SETUP_INFO, WIFI_VERSION_REQUEST)
@@ -66,10 +66,45 @@ RULES: tuple[Rule, ...] = (
)
@dataclass(frozen=True)
class BootstrapFix:
"""A Wi-Fi setup tweak for whichever receiver is chosen, run only when its experimental toggle is on."""
id: str
label: str
experiment: str # the bootstrap experiment it switches on
evidence: Evidence
status: str = EXPERIMENTAL
BOOTSTRAP_FIXES: tuple[BootstrapFix, ...] = (
BootstrapFix(id="phone_ping", label="Dongle Wi-Fi setup: ping like a phone", experiment="phone_ping",
evidence=Evidence("A Mavlou dongle never sends its Wi-Fi start request to the comma. A phone pings it once a second after its version reply.",
("2026-11-03 phone capture", "2026-11-07 sessions 5, 19, 24"), None)),
)
def rule_ids() -> tuple[str, ...]:
return tuple(rule.id for rule in RULES)
def experiment_ids() -> tuple[str, ...]:
return rule_ids() + tuple(fix.id for fix in BOOTSTRAP_FIXES)
def experimental_entries() -> list[dict]:
"""Every experimental fix the settings page lists: the car rules, then the device tweaks."""
return ([{"id": rule.id, "label": rule.label, "evidence": rule.evidence.failure} for rule in RULES if rule.status == EXPERIMENTAL] +
[{"id": fix.id, "label": fix.label, "evidence": fix.evidence.failure} for fix in BOOTSTRAP_FIXES if fix.status == EXPERIMENTAL])
def bootstrap_experiment(config: dict) -> str:
"""The Wi-Fi setup experiment to add, or "". Needs the master switch, the experiments switch and its own toggle."""
if not (config.get("head_unit_fixes", True) and config.get("head_unit_experiments")):
return ""
chosen = config.get("head_unit_experiment_rules") or []
return next((fix.experiment for fix in BOOTSTRAP_FIXES if fix.id in chosen), "")
def validate_rules(rules: tuple[Rule, ...] = RULES) -> list[str]:
"""Problems that would let a rule match more than the one build it was written for."""
problems = []
@@ -518,8 +518,7 @@ class Supervisor:
status["head_unit_fixes"] = bool(self.config["head_unit_fixes"])
status["head_unit_experiments"] = bool(self.config["head_unit_experiments"])
status["head_unit_experiment_rules"] = list(self.config["head_unit_experiment_rules"])
status["experimental_fixes"] = [{"id": rule.id, "label": rule.label, "evidence": rule.evidence.failure}
for rule in head_unit_fixes.RULES if rule.status == head_unit_fixes.EXPERIMENTAL]
status["experimental_fixes"] = head_unit_fixes.experimental_entries()
status["fix_suggestion"] = self._fix_suggestion()
status["car_candidates"] = list(getattr(self, "_candidates", []))[:8] # paired Android Auto devices to choose between
status["new_setup"] = new_setup.enabled()
@@ -573,7 +572,7 @@ class Supervisor:
self.log("head_unit_experiments_selected", enabled=bool(enabled))
def set_head_unit_experiment_rule(self, rule_id: str, enabled: bool) -> bool:
if rule_id not in head_unit_fixes.rule_ids():
if rule_id not in head_unit_fixes.experiment_ids():
return False
with self._lock:
rules = set(self.config["head_unit_experiment_rules"])
@@ -1647,7 +1646,7 @@ class Supervisor:
rfcomm = self._connect_rfcomm(address, channel)
boot = WirelessBootstrap(rfcomm, self._bootstrap_log, device_serial=config["device_name"],
version_status=int(config.get("version_status", 0)),
experiment="all" if car_profiles.tuning(config)["bootstrap_experiments"] or strategy.experiments else "",
experiment=self._bootstrap_experiment(strategy),
modern_reply=strategy.modern_reply, channel_override=strategy.channel_type,
phone_identity=strategy.phone_identity,
phone_address=self._adapter_address() if strategy.phone_identity else "",
@@ -2092,6 +2091,14 @@ class Supervisor:
except Exception:
pass
def _bootstrap_experiment(self, strategy) -> str:
"""The Wi-Fi setup experiment for this attempt: today's choice, plus a phone-style ping when its toggle is on."""
base = "all" if car_profiles.tuning(self.config)["bootstrap_experiments"] or strategy.experiments else ""
extra = head_unit_fixes.bootstrap_experiment(self.config)
if not extra:
return base
return "all_with_ping" if base == "all" and extra == "phone_ping" else extra
def _choose_fixes(self, head_unit: dict) -> tuple[str, ...]:
"""Called once the car has described itself: the extra channels this car gets this session, or none."""
config = self.config
@@ -30,3 +30,17 @@ def test_default_bootstrap_sends_no_pings_of_its_own():
from openpilot.starpilot.system.starport.tests.test_starport import run_bootstrap
_, _, seen, _ = run_bootstrap(version_first=True)
assert bs.WIFI_PING_REQUEST not in seen["messages"]
def test_unofficial_dongle_profile_and_last_rung_do_not_ping():
assert "phone_ping" not in bs.EXPERIMENTS["all"][1]
def test_phone_ping_needs_every_switch():
from openpilot.starpilot.system.starport import head_unit_fixes as hf
on = {"head_unit_fixes": True, "head_unit_experiments": True, "head_unit_experiment_rules": ["phone_ping"]}
assert hf.bootstrap_experiment(on) == "phone_ping"
assert hf.bootstrap_experiment({**on, "head_unit_fixes": False}) == ""
assert hf.bootstrap_experiment({**on, "head_unit_experiments": False}) == ""
assert hf.bootstrap_experiment({**on, "head_unit_experiment_rules": []}) == ""
assert hf.bootstrap_experiment({"head_unit_fixes": True}) == ""