Add car type and profile configuration options for Android Auto setup

This commit is contained in:
dirwin31
2026-10-26 13:00:26 -07:00
parent 0037bdd836
commit dbc8e0ea39
11 changed files with 441 additions and 9 deletions
+15
View File
@@ -2273,6 +2273,21 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
self.json(403, {'error': 'Park before changing the connection type'})
return
command, arguments = 'set_connection', {'connection': payload['connection']}
elif (action == 'car_type' and set(payload) == {'action', 'carType'} and
payload['carType'] in ('builtin', 'adapter', 'usb', 'app')):
if not configuration_allowed():
self.json(403, {'error': 'Park before changing the setup type'})
return
command, arguments = 'set_car_type', {'car_type': payload['carType']}
elif action == 'car_profile' and set(payload) == {'action', 'profile'} and isinstance(payload['profile'], str):
from openpilot.starpilot.system.android_auto.car_profiles import valid_override
if not valid_override(payload['profile']):
self.json(400, {'error': 'Invalid car profile'})
return
if not configuration_allowed():
self.json(403, {'error': 'Park before changing car settings'})
return
command, arguments = 'set_car_profile', {'profile': payload['profile']}
elif action == 'select_receiver' and set(payload) == {'action', 'address'} and isinstance(payload['address'], str):
if not configuration_allowed():
self.json(403, {'error': 'Park before changing the car'})
@@ -73,6 +73,8 @@ def api(tmp_path):
self.running = False
self.auto_connect = False
self.connection = 'wireless'
self.car_type = 'builtin'
self.profile = ''
self.available = True
self.stop_error = False
self.stop_completes = True
@@ -82,7 +84,7 @@ def api(tmp_path):
if command == 'status':
return {'status': {'receiver_address': self.receiver, 'receiver_name': 'My car', 'state': 'streaming' if self.running else 'idle',
'running': self.running, 'auto_connect': self.auto_connect, 'pairing_ready': False,
'connection': self.connection}}
'connection': self.connection, 'car_type': self.car_type}}
if command == 'devices':
return {'devices': [{'address': 'AA:BB:CC:DD:EE:FF', 'name': 'Saved car', 'paired': True, 'android_auto': True}]}
if command == 'select_receiver':
@@ -91,6 +93,10 @@ def api(tmp_path):
self.auto_connect = kwargs['enabled']
if command == 'set_connection':
self.connection = kwargs['connection']
if command == 'set_car_type':
self.car_type = kwargs['car_type']
if command == 'set_car_profile':
self.profile = kwargs['profile']
if command == 'start':
self.running = True
if command == 'stop':
@@ -427,6 +433,29 @@ def test_wired_connection_requires_park_but_not_a_paired_receiver(api, connectio
assert code == 409 and result['error'] == 'Verify your Android Auto package first'
def test_setup_type_and_car_profile_need_park_and_valid_values(api):
request, _, state, client = api
code, _, headers = request('POST', '/api/auth/login', json.dumps({'password': 'password123'}),
content_type='application/json')
assert code == 200
cookie = headers['Set-Cookie'].split(';', 1)[0]
def post(body):
return request('POST', '/api/android-auto/control', json.dumps(body), cookie=cookie, content_type='application/json')
state['parked'] = False
assert post({'action': 'car_type', 'carType': 'adapter'})[0] == 403
assert post({'action': 'car_profile', 'profile': 'default'})[0] == 403
state['parked'] = True
code, result, _ = post({'action': 'car_type', 'carType': 'adapter'})
assert code == 200 and result['status']['car_type'] == 'adapter'
assert post({'action': 'car_type', 'carType': 'hovercraft'})[0] == 400
assert post({'action': 'car_type', 'carType': 'usb', 'extra': 1})[0] == 400
assert post({'action': 'car_profile', 'profile': 'default'})[0] == 200 and client.profile == 'default'
assert post({'action': 'car_profile', 'profile': ''})[0] == 200 and client.profile == ''
assert post({'action': 'car_profile', 'profile': 'not-a-profile'})[0] == 400
def test_local_enable_then_upload_bundle_preserves_binary_bytes(api):
import io
import zipfile
@@ -473,3 +473,33 @@ const audioTool = catalog.tools.find(tool => tool.path === "/android-auto").chil
assert.equal(audioTool.name, "Spotify and Phone Audio")
assert.match(AndroidAutoPage.template, /href="#\/android-auto\/audio"/)
assert.ok(app.includes(`SpotifyPage v-else-if="['/android-auto/audio', '/spotify'].includes(route.path)"`))
// Setup type: what is in the car decides which pairing steps appear.
const typed = page(setup)
assert.equal(typed.carType, 'builtin', 'built-in wireless is the default')
typed.runtime = { car_type: 'adapter' }
assert.equal(typed.adapterSetup, true)
assert.equal(typed.pairsHere, true)
assert.equal(typed.findLabel, 'adapter')
typed.runtime = { car_type: 'usb' }
assert.equal(typed.pairsHere, false, 'a USB cable needs no Bluetooth pairing')
typed.runtime = { running: false, connection: 'wired' }
assert.equal(typed.carType, 'usb', 'older daemons without car_type still follow the connection')
typed.runtime = { connection: 'headunitpad' }
assert.equal(typed.carType, 'app')
const typedCalls = []
typed.control = (action, extra) => typedCalls.push([action, extra])
typed.saveCarType(0, 'Wireless adapter')
typed.saveCarType(0, 'Not a type')
assert.deepEqual(typedCalls, [['car_type', { carType: 'adapter' }]])
assert.deepEqual(new Set(AndroidAutoPage.computed.carTypeRow.call(typed).choices), new Set(['Built-in wireless', 'Wireless adapter', 'USB cable', 'Tablet app']))
// A tuned car profile only shows up when one is detected or pinned.
const profiled = page(setup)
assert.equal(profiled.showProfile, false, 'standard cars see nothing')
profiled.runtime = { car_profile_label: 'Hyundai Palisade', car_profile_source: 'detected' }
assert.equal(profiled.showProfile, true)
profiled.runtime = { car_profile_label: '', car_profile_source: 'override' }
assert.equal(profiled.showProfile, true, 'a pinned standard profile can be undone')
assert.match(AndroidAutoPage.template, /What’s in your car\?|carTypeRow/)
assert.match(AndroidAutoPage.template, /runtime\?\.pairing_note/)
assert.match(AndroidAutoPage.template, /'Find ' \+ findLabel/)
console.log('Android Auto page: setup type, adapter checklist and car profile line passed')
@@ -66,6 +66,9 @@ const APP_SETUP = `
<li>Tap Connect here, then tap the WiFi button in Open Headunit.</li>
</ol>
</div>`
export const CAR_TYPES = {
builtin: "Built-in wireless", adapter: "Wireless adapter", usb: "USB cable", app: "Tablet app"
}
const PROMPT = new Set(["pin", "passkey", "confirmation", "authorization", "display_pin", "display_passkey"])
export function validPromptInput(prompt, value) {
@@ -283,7 +286,7 @@ export class AndroidAutoFeed {
async control(action, extra = {}) {
if (!this.active || this.busy || !this.setup?.enabled || this.pairing?.active ||
!["start", "stop", "select_receiver", "auto_connect", "connection"].includes(action)) return false
!["start", "stop", "select_receiver", "auto_connect", "connection", "car_type", "car_profile"].includes(action)) return false
const result = await this.request("./api/android-auto/control", {
method: "POST",
headers: { "Content-Type": "application/json" }, body: JSON.stringify({ action, ...extra })
@@ -496,6 +499,15 @@ export const AndroidAutoPage = {
return { label: "App", value: this.app || "Off", choices: ["Off", ...Object.values(HEAD_UNIT_APPS)], available: true, action: true,
reason: "Project to a tablet app on the same Wi-Fi as the comma or the Galaxy Wi-Fi Hotspot instead of the car." }
},
carType() { return this.runtime?.car_type || (this.wired ? "usb" : this.app ? "app" : "builtin") },
adapterSetup() { return this.carType === "adapter" },
pairsHere() { return this.carType === "builtin" || this.carType === "adapter" },
findLabel() { return this.adapterSetup ? "adapter" : "car" },
carTypeRow() {
return { label: "What’s in your car?", value: CAR_TYPES[this.carType], choices: Object.values(CAR_TYPES), available: true, action: true,
reason: "Built-in: the car supports wireless Android Auto. Adapter: a dongle such as AAWireless, Motorola MA1 or Carlinkit. Both the adapter and your car get paired." }
},
showProfile() { return !!(this.runtime?.car_profile_label || this.runtime?.car_profile_source === "override") },
appDisabled() {
return !this.setup?.enabled || !this.setup.parked || !!this.pairing?.active || !!this.runtime?.running
},
@@ -580,6 +592,10 @@ export const AndroidAutoPage = {
},
methods: {
switchRow,
saveCarType(index, value) {
const carType = Object.keys(CAR_TYPES).find(key => CAR_TYPES[key] === value)
return carType ? this.control("car_type", { carType }) : false
},
saveApp(index, value) {
const connection = Object.keys(HEAD_UNIT_APPS).find(key => HEAD_UNIT_APPS[key] === value) || "wireless"
return this.control("connection", { connection })
@@ -763,9 +779,32 @@ export const AndroidAutoPage = {
<li class="gx-card gx-aa-step" :class="{ 'gx-aa-step--done': !!selected && !pairing?.active }">
<div class="gx-aa-step__head">
<span class="gx-aa-step__num" aria-hidden="true"><i v-if="selected && !pairing?.active" class="bi bi-check-lg"></i><template v-else>2</template></span>
<div><h3>Pair Your Car</h3><p>Open your car’s phone pairing screen, then tap Find car. You can do this before step 1 is finished.</p></div>
<div><h3>{{ adapterSetup ? 'Pair Your Adapter and Car' : 'Pair Your Car' }}</h3><p>{{ adapterSetup ? 'An adapter needs two pairings: the adapter here, and your car on the Bluetooth page.' : pairsHere ? 'Open your car’s phone pairing screen, then tap Find car. You can do this before step 1 is finished.' : 'No Bluetooth pairing is needed for this setup.' }}</p></div>
</div>
<div class="gx-aa-step__body">
<GalaxySettingRow :row="carTypeRow" :index="0" :busy="busy" :disabled="appDisabled" :save-value="saveCarType" />
<p v-if="showProfile" class="gx-note" role="status">
{{ runtime.car_profile_source === 'override' ? 'Using standard settings for this car.' : 'Tuned for ' + runtime.car_profile_label + ' (detected).' }}
<button type="button" class="gx-btn gx-btn--tonal" :disabled="busy || !setup.parked"
@click="control('car_profile', { profile: runtime.car_profile_source === 'override' ? '' : 'default' })">{{ runtime.car_profile_source === 'override' ? 'Use detected settings' : 'Use standard settings' }}</button></p>
<p v-if="carType === 'usb'" class="gx-note">Plug the car’s data USB port into the comma’s USB-C port, then turn on Wired USB Connection in step 3.</p>
<p v-else-if="carType === 'app'" class="gx-note">Pick your tablet app in step 3 and follow its setup steps.</p>
<template v-if="pairsHere">
<ol v-if="adapterSetup" class="gx-aa-list" aria-label="Pairings needed">
<li class="gx-aa-row">
<i class="bi bi-usb-symbol" aria-hidden="true"></i>
<span class="gx-aa-row__text"><strong>Adapter</strong><small>{{ selected ? selected.name : 'Put it in pairing mode, then tap Find adapter.' }}</small></span>
<span class="gx-aa-pill" :class="selected ? 'gx-aa-pill--ready' : ''">{{ selected ? 'Paired' : 'Not paired' }}</span>
</li>
<li class="gx-aa-row">
<i class="bi bi-car-front" aria-hidden="true"></i>
<span class="gx-aa-row__text"><strong>Your car</strong><small>{{ runtime?.companion_name ? runtime.companion_name : selected ? 'Pair your car on the Bluetooth page, then tap Connect there.' : 'Pair the adapter first.' }}</small></span>
<span v-if="runtime?.companion_name" class="gx-aa-pill gx-aa-pill--ready">Paired</span>
<a v-else-if="selected" class="gx-btn gx-btn--tonal" href="#/bluetooth">Open Bluetooth</a>
<span v-else class="gx-aa-pill">Not paired</span>
</li>
</ol>
<GxNotice v-if="!pairing?.active && runtime?.pairing_note" tone="warn">{{ runtime.pairing_note }}</GxNotice>
<ul v-if="!pairing?.active && receivers.length" class="gx-aa-list" aria-label="Paired cars">
<li v-for="car in receivers" :key="car.address" class="gx-aa-row">
<i class="bi bi-car-front" aria-hidden="true"></i>
@@ -806,14 +845,14 @@ export const AndroidAutoPage = {
<p v-if="pairWhenReady" class="gx-note" role="status"><span class="gx-spinner" aria-hidden="true"></span> Turning on Android Auto, then searching…</p>
<p v-else-if="!pairing?.active && pairingReason" class="gx-note" role="status">{{ pairingReason }}</p>
<div class="gx-driving__actions gx-actions">
<button v-if="!pairing?.active" class="gx-btn" :class="{ 'gx-btn--tonal': !!selected }" :disabled="pairWhenReady || !!pairingReason" @click="startPairing"><i class="bi bi-search"></i> {{ selected ? 'Pair another' : 'Find car' }}</button>
<button v-if="!pairing?.active" class="gx-btn" :class="{ 'gx-btn--tonal': !!selected }" :disabled="pairWhenReady || !!pairingReason" @click="startPairing"><i class="bi bi-search"></i> {{ selected ? 'Pair another ' + findLabel : 'Find ' + findLabel }}</button>
<button v-else class="gx-btn gx-btn--tonal" @click="cancelPairing">Cancel Search / Pairing</button>
</div>
<details class="gx-aa-wiki"><summary>Stuck pairing? Pairing order and wireless adapters</summary>
<ul>
<li><strong>Keep this page open while pairing</strong> so the car’s PIN or code doesn’t time out.</li>
<li><strong>Cars with built-in wireless Android Auto:</strong> open the car’s phone pairing screen, then tap <strong>Find car</strong> above. Choose your car, then confirm the code on both screens.</li>
<li><strong>Wireless adapter</strong> (AAWireless, Motorola MA1, Carlinkit…): pair your car in Bluetooth under Tools first so calls keep working, then put the adapter in pairing mode and pair it here. Adapter names often look like AndroidAuto-XXXX.</li>
<li><strong>Wireless adapter</strong> (AAWireless, Motorola MA1, Carlinkit…): choose Wireless adapter above, put the adapter in pairing mode and pair it here, and pair your car in Bluetooth under Tools so calls keep working. Then connect the car there. Adapter names often look like AndroidAuto-XXXX.</li>
</ul>
<p><strong>Start fresh:</strong></p>
<ol>
@@ -824,6 +863,7 @@ export const AndroidAutoPage = {
</ol>
<a class="gx-btn gx-btn--tonal" href="#/bluetooth">Open Bluetooth</a>
</details>
</template>
</div>
</li>
@@ -217,6 +217,7 @@ export const BluetoothPage = {
<template v-else>
<GxNotice tone="danger" v-if="error">{{ error }}</GxNotice>
<GxState v-if="!status && !error" loading>Checking Bluetooth…</GxState>
<GxNotice v-if="status" tone="info">Using a wireless Android Auto adapter? Pair the adapter on the <a class="gx-link" href="#/android-auto">Android Auto page</a>, then pair and connect your car here.</GxNotice>
<template v-else-if="status">
<section class="gx-card"><div class="gx-section__header"><i class="bi bi-bluetooth" aria-hidden="true"></i><span class="gx-section__title">Adapter</span></div>
<div class="gx-panel gx-stack">
@@ -387,6 +387,11 @@ class PairingAttempt:
self.record["steps"].pop(0)
self.record["steps_omitted"] = self.record.get("steps_omitted", 0) + 1
def reached_connection(self) -> bool:
"""The car was approved or connected before the window ended."""
return any(step.get("step") == "status" and (step.get("approved") or step.get("state") in ("paired", "connected"))
for step in self.record["steps"])
def finish(self, result: str, **values) -> None:
self.step(result, **values)
self._sink({**self.record, "result": result, "duration_s": round(self._clock() - self._opened, 1)})
@@ -0,0 +1,61 @@
"""Per-car tuning, picked by fingerprint so nobody has to choose their car from a list.
A car is recognised from what its head unit reported on a previous connection. Cars without a rule
keep the standard behaviour exactly. Add a rule to ship a variation for one car; ``car_profile_override``
in the config pins a profile (or "default") for a single comma while testing with someone.
"""
import re
DEFAULTS = {"connect_delay_s": 0.0} # wait before the Bluetooth connect, for head units that need time to wake
FINGERPRINT_KEYS = ("head_unit_make", "head_unit_model", "car_make", "car_model", "car_year")
# {"id": "...", "label": "...", "match": {fingerprint key: regex, ...}, "tuning": {key from DEFAULTS: value}}
RULES: list[dict] = []
def profile_ids() -> list[str]:
return [rule["id"] for rule in RULES]
def valid_override(value: str) -> bool:
return value in ("", "default", *profile_ids())
def fingerprint_from(head_unit: dict, address: str) -> dict:
found = {key: str(head_unit[key])[:80] for key in FINGERPRINT_KEYS if head_unit.get(key)}
return {"address": address.upper(), **found} if found else {}
def _matches(rule: dict, fingerprint: dict) -> bool:
match = rule.get("match") or {}
return bool(match) and all(re.search(pattern, str(fingerprint.get(key, "")), re.IGNORECASE) for key, pattern in match.items())
def resolve(config: dict) -> tuple[str, str]:
"""(profile id, source) where source is "override", "detected" or "standard"."""
override = config.get("car_profile_override", "")
if override == "default":
return "default", "override"
if override in profile_ids():
return override, "override"
fingerprint = config.get("car_fingerprint") or {}
if fingerprint.get("address") == str(config.get("receiver_address", "")).upper():
for rule in RULES:
if _matches(rule, fingerprint):
return rule["id"], "detected"
return "default", "standard"
def tuning(config: dict) -> dict:
profile, _ = resolve(config)
rule = next((item for item in RULES if item["id"] == profile), {})
return {**DEFAULTS, **{key: value for key, value in (rule.get("tuning") or {}).items() if key in DEFAULTS}}
def summary(config: dict) -> dict:
"""What the UI may show: nothing at all while the standard behaviour applies."""
profile, source = resolve(config)
rule = next((item for item in RULES if item["id"] == profile), None)
return {"car_profile": profile, "car_profile_source": source, "car_profile_label": rule["label"] if rule else "",
"car_profile_ids": profile_ids()}
@@ -28,7 +28,7 @@ from openpilot.starpilot.system.android_auto.source_verifier import GalaxySource
from openpilot.starpilot.system.android_auto.supervisor import Supervisor
PAIRING_COMMANDS = {"prepare_pairing", "pairing_status", "pairing_response", "cancel_pairing", "pair_device"}
AUTHENTICATED_COMMANDS = {"start", "set_auto_connect", "select_receiver", "set_view", "set_connection",
AUTHENTICATED_COMMANDS = {"start", "set_auto_connect", "select_receiver", "set_view", "set_connection", "set_car_type", "set_car_profile",
"prepare_pairing", "pairing_status", "pairing_response", "cancel_pairing", "pair_device", "devices", "bluetooth_action",
"phone_window", "bluetooth_power"}
@@ -124,6 +124,10 @@ def handle(supervisor: Supervisor, request: dict[str, Any], verifier: GalaxySour
supervisor.set_view(str(request.get("view", "")))
elif command == "set_connection":
supervisor.set_connection(str(request.get("connection", "")))
elif command == "set_car_type":
supervisor.set_car_type(str(request.get("car_type", "")))
elif command == "set_car_profile":
supervisor.set_car_profile(str(request.get("profile", "")))
elif command == "prepare_pairing":
supervisor.prepare_pairing()
elif command == "pairing_status":
@@ -131,6 +131,9 @@ DEFAULT_CONFIG = {
"phone_class": True, # while pairing/connecting, present as a phone: HFP gateway + smartphone Class of Device
"auto_connect": True, # start projection on its own when the chosen car is on (onroad, or it reaches the comma)
"rfcomm_cache": {}, # car address -> Android Auto RFCOMM channel learned over SDP, to skip discovery next time
"car_type": "", # wireless setup shown in Galaxy: "builtin" or "adapter"; "" = work it out from the saved companion car
"car_profile_override": "", # "" = pick by fingerprint; "default" = standard behaviour; otherwise a profile id from car_profiles
"car_fingerprint": {}, # what the chosen car's head unit reported last time, used to pick a profile before it connects
}
@@ -160,6 +163,9 @@ def load_config(path: Path | None = None) -> dict:
config["connection"] = "wireless"
if not 1024 <= int(config["headunitpad_port"]) <= 65535:
config["headunitpad_port"] = DEFAULT_CONFIG["headunitpad_port"]
if config["car_type"] not in ("", "builtin", "adapter"):
config["car_type"] = ""
config["car_fingerprint"] = {str(k): str(v)[:80] for k, v in config["car_fingerprint"].items()}
if config["usb_mode"] not in ("auto", "handshake", "direct"):
config["usb_mode"] = "auto"
config["bitrate_kbps"] = max(1000, min(12000, int(config["bitrate_kbps"])))
@@ -32,7 +32,7 @@ from pathlib import Path
from typing import Any
from openpilot.common.hardware import COMMA_HARDWARE
from openpilot.starpilot.system.android_auto import identity as identity_store
from openpilot.starpilot.system.android_auto import car_profiles, identity as identity_store
from openpilot.starpilot.system.android_auto.auto_connect import AutoConnectPolicy
from openpilot.starpilot.system.android_auto.bootstrap import NAMES, BootstrapError, WirelessBootstrap
from openpilot.starpilot.system.android_auto.bugsink_report import PairingAttempt
@@ -59,6 +59,7 @@ ENCODER_RECOVERIES = 3 # hardware encoder reopens allowed per window befor
ENCODER_RECOVERY_WINDOW = 10.0
CAR_LINK_HOLD = 10.0 # a hands-free connection from the car counts as "car present" this long
HFP_FRESH = 3.0 # a hands-free link from the car this recent still means "the car is reaching out now"
NO_ANDROID_AUTO_WAIT = 8.0 # a paired device must offer Android Auto within this long, or we say it does not
USB_HANDSHAKE_WAIT = 10.0 # the car connected but sent no AOA START this long -> present as an accessory directly
USB_CONFIGURE_WAIT = 15.0 # slower receivers need time to enumerate after the AOA switch
LINK_CHECK_INTERVAL = 1.0 # how often the link watcher asks whether the car's Wi-Fi is still up
@@ -286,6 +287,8 @@ class EventLog:
class Supervisor:
_pairing_note = "" # why a paired car was not chosen, kept after the pairing window closes
_no_aa_since = 0.0
def __init__(self, bluez_factory=None, lease_factory=None, bluetooth_client=None, frame_path: str | None = None,
synthetic: bool = False, car_frame_path: str = CAR_FRAME_PATH, touch_path: str = DEFAULT_TOUCH_SOCKET,
control_path: str = DEFAULT_CONTROL_SOCKET, renderer_command: list[str] | None = None, onroad=_is_onroad,
@@ -347,6 +350,9 @@ class Supervisor:
status["receiver_name"] = self.config["receiver_name"]
status["companion_address"] = self.config.get("companion_address", "")
status["companion_name"] = self.config.get("companion_name", "")
status["pairing_note"] = self._pairing_note
status["car_type"] = self.car_type()
status.update(car_profiles.summary(self.config))
status["configured_view"] = self.config["view"]
status["connection"] = self.config["connection"]
status["wired_blocked"] = wired_blocker()
@@ -430,7 +436,7 @@ class Supervisor:
if self._bluez is not None and self.config['receiver_address'] and self.config['receiver_address'] != address:
self._bluez.release()
if self.config["receiver_address"] != address:
self.config.update(companion_address="", companion_name="")
self.config.update(companion_address="", companion_name="", car_fingerprint={})
self.config.update(receiver_address=address, receiver_name=name or address)
identity_store.save_config(self.config)
self._unselect_car_audio(address)
@@ -458,7 +464,7 @@ class Supervisor:
if chosen:
if self._bluez is not None:
self._bluez.release()
self.config.update(receiver_address="", receiver_name="", companion_address="", companion_name="")
self.config.update(receiver_address="", receiver_name="", companion_address="", companion_name="", car_fingerprint={})
elif companion:
self.config.update(companion_address="", companion_name="")
identity_store.save_config(self.config)
@@ -487,6 +493,44 @@ class Supervisor:
identity_store.save_config(self.config)
self.log("connection_selected", connection=connection)
def car_type(self) -> str:
"""What is in the car, for the setup page: wired and tablet follow the connection, wireless is built-in or an adapter."""
connection = self.config["connection"]
if connection == "wired":
return "usb"
if connection in ("headunitpad", "openheadunit"):
return "app"
return self.config["car_type"] or ("adapter" if self.config.get("companion_address") else "builtin")
def set_car_type(self, car_type: str) -> None:
if car_type not in ("builtin", "adapter", "usb", "app"):
raise RuntimeError(f"Unknown car type {car_type!r}")
connection = {"usb": "wired", "app": "headunitpad" if self.config["connection"] not in ("headunitpad", "openheadunit")
else self.config["connection"]}.get(car_type, "wireless")
if connection != self.config["connection"]:
self.set_connection(connection)
if car_type in ("builtin", "adapter"):
with self._lock:
self.config["car_type"] = car_type
identity_store.save_config(self.config)
self.log("car_type_selected", car_type=car_type)
def set_car_profile(self, profile: str) -> None:
"""Pin a car profile ("default" for standard behaviour) or go back to picking it by fingerprint ("")."""
if not car_profiles.valid_override(profile):
raise RuntimeError(f"Unknown car profile {profile!r}")
with self._lock:
self.config["car_profile_override"] = profile
identity_store.save_config(self.config)
self.log("car_profile_selected", profile=profile or "auto")
def _remember_fingerprint(self, head_unit: dict) -> None:
fingerprint = car_profiles.fingerprint_from(head_unit, self.config["receiver_address"])
if fingerprint and fingerprint != self.config.get("car_fingerprint"):
with self._lock:
self.config["car_fingerprint"] = fingerprint
identity_store.save_config(self.config)
def prepare_pairing(self, seconds: float = 180.0) -> None:
"""Present as a phone (HFP gateway, smartphone class) while the car pairs."""
if self._bluez_factory is None and (self._source_session is None or self._shared_bluetooth_owner is None or
@@ -506,6 +550,7 @@ class Supervisor:
self._pairing_known = known
self._pairing_until = time.monotonic() + seconds
self._pairing_attempt, self._pairing_seen = PairingAttempt(seconds), ()
self._pairing_note, self._no_aa_since = "", 0.0
self.log("pairing_window", seconds=seconds)
def pair_device(self, address: str) -> None:
@@ -533,6 +578,8 @@ class Supervisor:
with self._lock:
attempt, self._pairing_attempt = getattr(self, '_pairing_attempt', None), None
if attempt is not None:
if result in ("interrupted", "cancelled", "expired") and attempt.reached_connection():
values["ended"], result = result, "paired_not_selected"
attempt.finish(result, **values)
def pairing_response(self, prompt_id: str, accepted: bool, value: str = '') -> None:
@@ -715,6 +762,7 @@ class Supervisor:
self._shared_bluetooth_owner.phone_lease is not None and
self._shared_bluetooth_owner.phone_lease.receiver == device["address"])]
if not new:
self._note_missing_android_auto(devices, receiver)
return
with self._lock:
attempt, self._pairing_attempt = getattr(self, '_pairing_attempt', None), None
@@ -729,6 +777,22 @@ class Supervisor:
if attempt is not None:
attempt.finish("select_failed", error=str(error))
def _note_missing_android_auto(self, devices: list, receiver: dict | None) -> None:
"""An approved, paired device that still does not offer wireless Android Auto after a short wait is not a car we can use."""
device = next((item for item in devices if receiver is not None and item["address"] == receiver["address"] and
item["paired"] and not item.get("android_auto")), None)
if device is None:
self._no_aa_since = 0.0
return
now = time.monotonic()
if not self._no_aa_since:
self._no_aa_since = now
elif now - self._no_aa_since >= NO_ANDROID_AUTO_WAIT and not self._pairing_note:
self._pairing_note = (f"{device['name'] or 'This device'} paired, but it doesn't offer wireless Android Auto. "
"Using a wireless adapter? Choose Wireless adapter, pair the adapter here, then connect your car on the Bluetooth page.")
self._pairing_step("no_android_auto")
self.log("pairing_no_android_auto", address=device["address"])
def close(self) -> None:
self.stop()
if self._bluez is not None:
@@ -1051,6 +1115,9 @@ class Supervisor:
self.log('companion_car_connected', address=companion)
elif device.get('android_auto') and '0000111e-0000-1000-8000-00805f9b34fb' not in device.get('uuids', []):
raise RuntimeError('Connect your paired car in Galaxy Bluetooth first to select it for this Android Auto adapter')
if delay := car_profiles.tuning(config)["connect_delay_s"]:
self.log("car_profile_connect_delay", seconds=delay)
self._wait(delay)
bluez.connect_device(address)
self._wait(1.0)
@@ -1079,6 +1146,7 @@ class Supervisor:
if source == "sdp":
self._remember_channel(address, channel)
self._set(head_unit=result.head_unit)
self._remember_fingerprint(result.head_unit)
bluez.restore_class() # the car has accepted the comma; stop looking like a phone to everything else
keepalive_stop = threading.Event()
threading.Thread(target=boot.keepalive, args=(keepalive_stop,), name="aa_rfcomm_keepalive", daemon=True).start()
@@ -0,0 +1,173 @@
"""Car profiles are picked by fingerprint; cars without a rule keep the standard behaviour."""
import json
import threading
from unittest.mock import Mock
import pytest
from openpilot.starpilot.system.android_auto import bugsink_report as br, car_profiles, daemon, identity as identity_store
from openpilot.starpilot.system.android_auto.supervisor import Supervisor
from openpilot.starpilot.system.android_auto.tests.fake_head_unit import make_identity
from openpilot.starpilot.system.android_auto.tests.test_android_auto import make_supervisor
ADAPTER = "AA:BB:CC:DD:EE:01"
PALISADE = {"id": "palisade", "label": "Hyundai Palisade", "match": {"car_make": "^hyundai$", "car_model": "palisade"},
"tuning": {"connect_delay_s": 10.0, "unknown_key": 1}}
@pytest.fixture(scope="module")
def identity(tmp_path_factory):
return make_identity(tmp_path_factory.mktemp("identity"))
@pytest.fixture
def sup(identity, tmp_path, monkeypatch):
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: None)
return sup
@pytest.fixture
def palisade(monkeypatch):
monkeypatch.setattr(car_profiles, "RULES", [PALISADE])
def config(**values):
return {"receiver_address": ADAPTER, "car_profile_override": "", "car_fingerprint": {}, **values}
def test_every_car_is_standard_without_rules():
assert car_profiles.resolve(config(car_fingerprint={"address": ADAPTER, "car_make": "Hyundai"})) == ("default", "standard")
assert car_profiles.tuning(config()) == car_profiles.DEFAULTS
assert car_profiles.summary(config())["car_profile_label"] == ""
def test_fingerprint_picks_a_profile_only_for_the_chosen_receiver(palisade):
fingerprint = {"address": ADAPTER, "car_make": "Hyundai", "car_model": "Palisade"}
assert car_profiles.resolve(config(car_fingerprint=fingerprint)) == ("palisade", "detected")
assert car_profiles.resolve(config(car_fingerprint={**fingerprint, "address": "AA:BB:CC:DD:EE:09"})) == ("default", "standard")
assert car_profiles.resolve(config(car_fingerprint={"address": ADAPTER, "car_make": "Hyundai", "car_model": "Sonata"})) == ("default", "standard")
def test_tuning_keeps_only_known_keys(palisade):
tuned = car_profiles.tuning(config(car_profile_override="palisade"))
assert tuned == {"connect_delay_s": 10.0}
def test_override_pins_a_profile_or_forces_standard(palisade):
fingerprint = {"address": ADAPTER, "car_make": "Hyundai", "car_model": "Palisade"}
assert car_profiles.resolve(config(car_profile_override="default", car_fingerprint=fingerprint)) == ("default", "override")
assert car_profiles.resolve(config(car_profile_override="palisade")) == ("palisade", "override")
assert car_profiles.resolve(config(car_profile_override="gone")) == ("default", "standard")
assert car_profiles.valid_override("") and car_profiles.valid_override("default") and car_profiles.valid_override("palisade")
assert not car_profiles.valid_override("gone")
def test_old_config_files_load_with_standard_defaults(tmp_path):
path = tmp_path / "config.json"
path.write_text(json.dumps({"config_version": 6, "connection": "wired", "car_type": "bogus"}))
loaded = identity_store.load_config(path)
assert loaded["car_type"] == "" and loaded["car_profile_override"] == "" and loaded["car_fingerprint"] == {}
assert loaded["connection"] == "wired"
def test_car_type_follows_the_connection_and_companion(sup):
assert sup.status()["car_type"] == "builtin"
sup.config.update(companion_address="car")
assert sup.car_type() == "adapter"
sup.config.update(car_type="builtin")
assert sup.car_type() == "builtin"
sup.config.update(connection="headunitpad")
assert sup.car_type() == "app"
def test_choosing_a_car_type_sets_the_connection(sup):
sup.set_car_type("adapter")
assert sup.car_type() == "adapter" and sup.config["connection"] == "wireless"
sup.set_car_type("app")
assert sup.config["connection"] == "headunitpad" and sup.car_type() == "app"
sup.set_car_type("builtin")
assert sup.config["connection"] == "wireless" and sup.car_type() == "builtin"
assert json.loads(identity_store.CONFIG_PATH.read_text())["car_type"] == "builtin"
with pytest.raises(RuntimeError):
sup.set_car_type("hovercraft")
def test_profile_changes_are_pinned_and_validated(sup, palisade):
sup.set_car_profile("palisade")
status = sup.status()
assert status["car_profile"] == "palisade" and status["car_profile_source"] == "override" and status["car_profile_label"] == "Hyundai Palisade"
sup.set_car_profile("")
assert sup.status()["car_profile_source"] == "standard"
with pytest.raises(RuntimeError):
sup.set_car_profile("nope")
def test_setup_commands_need_a_signed_in_galaxy_source():
assert {"set_car_type", "set_car_profile"} <= daemon.AUTHENTICATED_COMMANDS
def test_head_unit_fingerprint_is_remembered_and_dropped_with_the_car(sup):
sup.config["receiver_address"] = ADAPTER
sup._remember_fingerprint({"car_make": "Hyundai", "car_model": "Palisade", "head_unit_make": "MOBIS", "ignored": "x"})
stored = json.loads(identity_store.CONFIG_PATH.read_text())["car_fingerprint"]
assert stored == {"address": ADAPTER, "head_unit_make": "MOBIS", "car_make": "Hyundai", "car_model": "Palisade"}
sup._remember_fingerprint({})
assert sup.config["car_fingerprint"] == stored
def make_notifier():
sup = Supervisor.__new__(Supervisor)
sup._lock = threading.RLock()
sup._pairing_step = Mock()
sup.log = Mock()
return sup
def test_a_paired_device_without_android_auto_is_explained_after_a_wait(monkeypatch):
sup = make_notifier()
receiver = {"address": ADAPTER, "name": "Civic"}
devices = [{"address": ADAPTER, "name": "Civic", "paired": True, "android_auto": False}]
clock = iter([100.0, 103.0, 109.0])
monkeypatch.setattr("time.monotonic", lambda: next(clock))
for _ in range(2):
sup._note_missing_android_auto(devices, receiver)
assert sup._pairing_note == ""
sup._note_missing_android_auto(devices, receiver)
assert "doesn't offer wireless Android Auto" in sup._pairing_note and "Wireless adapter" in sup._pairing_note
sup._pairing_step.assert_called_once_with("no_android_auto")
def test_a_device_that_offers_android_auto_never_gets_a_note():
sup = make_notifier()
sup._no_aa_since = 5.0
sup._note_missing_android_auto([{"address": ADAPTER, "name": "Car", "paired": True, "android_auto": True}], {"address": ADAPTER})
assert sup._pairing_note == "" and sup._no_aa_since == 0.0
@pytest.mark.parametrize("end", ["interrupted", "cancelled", "expired"])
def test_a_pairing_that_connected_is_not_reported_as_interrupted(end):
records = []
attempt = br.PairingAttempt(180, sink=records.append)
attempt.step("status", state="connected", prompt=False, approved=True)
sup = make_notifier()
sup._pairing_attempt = attempt
sup._end_pairing(end, reason="Bluetooth lease released")
assert records[0]["result"] == "paired_not_selected" and records[0]["steps"][-1]["ended"] == end
def test_a_pairing_that_never_connected_keeps_its_result():
records = []
attempt = br.PairingAttempt(180, sink=records.append)
attempt.step("status", state="idle", prompt=False, approved=False)
sup = make_notifier()
sup._pairing_attempt = attempt
sup._end_pairing("interrupted", reason="Galaxy session ended")
assert records[0]["result"] == "interrupted"
def test_the_new_result_has_a_bugsink_report():
attempt = br.PairingAttempt(180, sink=lambda record: None)
attempt.step("status", state="connected", prompt=False, approved=True)
event = br.pairing_event({**attempt.record, "result": "paired_not_selected", "duration_s": 5.0})
assert event["fingerprint"] == ["android-auto-pairing", "paired_not_selected"] and event["level"] == "warning"