Android Auto Connection and Rendering

This commit is contained in:
dirwin31
2026-10-06 00:41:22 -05:00
committed by firestar5683
parent 875eb6bc8c
commit acffe8f112
30 changed files with 2526 additions and 105 deletions
@@ -0,0 +1,80 @@
"""Read-only access to the Android Auto logs under /data/android_auto/logs, for Logs & Diagnostics.
Lists the kept session logs with a one-line summary each, and serves one log or the shareable
bundle (``compat_report.bundle``: every session with its report, the settings without Bluetooth
addresses, and the renderer logs). Only names the log directory itself lists are served.
"""
from __future__ import annotations
from collections.abc import Sequence
from datetime import datetime
from pathlib import Path
from openpilot.starpilot.system.android_auto import compat_report
from openpilot.starpilot.system.android_auto import identity as identity_store
MAX_FILE_BYTES = 8 << 20 # a session log is normally well under 1 MiB; the renderer logs rotate at 256 KiB
class LogsUnavailable(Exception):
pass
class LogMissing(Exception):
pass
def _car(report: dict) -> str:
car = report.get("car") or {}
make = car.get("car_make") or car.get("head_unit_make") or ""
model = car.get("car_model") or car.get("head_unit_model") or ""
return " ".join(part for part in (make, model) if part)
class AndroidAutoLogs:
def __init__(self, directory: Path | None = None, config_path: Path | None = None):
self.directory = directory if directory is not None else identity_store.LOG_DIR
self.config_path = config_path
def _files(self) -> Sequence[Path]:
if not self.directory.is_dir():
return []
extras = [self.directory / name for name in compat_report.EXTRA_LOGS]
return compat_report.session_logs(self.directory) + [path for path in extras if path.is_file() and not path.is_symlink()]
def list(self) -> dict:
try:
sessions, others = [], []
for path in self._files():
info = path.stat()
item = {"name": path.name, "size": info.st_size, "modifiedAt": int(info.st_mtime)}
if path.name.startswith("session-"):
report = compat_report.summarize(compat_report.load_events(path))
item.update(outcome=report.get("outcome") or "", started=report.get("started") or "",
transport=report.get("transport") or "", trigger=report.get("trigger") or "", car=_car(report))
sessions.append(item)
else:
others.append(item)
except OSError:
raise LogsUnavailable from None
return {"schemaVersion": 1, "sessions": sessions, "others": others}
def file(self, name: str) -> tuple[str, bytes]:
path = next((path for path in self._files() if path.name == name), None)
if path is None or path.is_symlink():
raise LogMissing
try:
return path.name, compat_report.read_log(path, MAX_FILE_BYTES)
except FileNotFoundError:
raise LogMissing from None
except OSError:
raise LogsUnavailable from None
def bundle(self, now: datetime | None = None) -> tuple[str, bytes]:
try:
data = compat_report.bundle(self.directory, self.config_path)
except OSError:
raise LogsUnavailable from None
stamp = (now or datetime.now()).strftime("%Y%m%d-%H%M%S")
return f"starpilot-android-auto-logs-{stamp}.zip", data
+53 -1
View File
@@ -179,7 +179,7 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
sounds=None, software_operations=None, drive_stats=None, layout_preview_socket=None, controllers_socket=None,
remote_pairing=None, parked=None, camera_snapshot=None, clock=time.monotonic, android_auto_setup=None,
android_auto_client=None, navigation=None, drive_state=None, cloud_provider=None, cloud_offroad=None, projection_layout=None,
local_access=None, tmux_live=None):
local_access=None, tmux_live=None, android_auto_logs=None):
cloud = cloud_provider
def cloud_status():
@@ -194,6 +194,14 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
console_source = tmux_live if tmux_live is not None else TmuxLive()
monitor_source = monitor if monitor is not None else SystemMonitor()
reports = crashes if crashes is not None else CrashReports()
aa_logs_source = android_auto_logs
def aa_logs():
nonlocal aa_logs_source
if aa_logs_source is None:
from openpilot.starpilot.galaxy.android_auto_logs import AndroidAutoLogs
aa_logs_source = AndroidAutoLogs()
return aa_logs_source
software_source = software
software_operations_source = software_operations
software_lock = threading.Lock()
@@ -561,6 +569,19 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
def json(self, status, value):
self.respond(status, json.dumps(value, allow_nan=False).encode())
def attachment(self, filename, body, content_type):
"""A download the browser saves as ``filename`` rather than displays."""
self.send_response(200)
self.send_header('Content-Type', content_type)
self.send_header('Content-Length', str(len(body)))
self.send_header('Content-Disposition', f'attachment; filename="{filename}"')
self.send_header('Cache-Control', 'no-store')
self.send_header('X-Content-Type-Options', 'nosniff')
self.send_header('Referrer-Policy', 'no-referrer')
self.end_headers()
if self.command != 'HEAD':
self.wfile.write(body)
def camera_video(self, encoded_name: str, *, combined=False):
if not self.require_session():
return
@@ -867,6 +888,37 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
# immediately before sending potentially sensitive content.
if self.require_session():
self.json(200, result)
elif path == '/api/android-auto/logs' or path.startswith('/api/android-auto/logs/'):
if not self.require_session():
return
from openpilot.starpilot.galaxy.android_auto_logs import LogMissing, LogsUnavailable
name = unquote(path.removeprefix('/api/android-auto/logs/file/')) if path.startswith('/api/android-auto/logs/file/') else None
if name is not None and not re.fullmatch(r'[\w.-]{1,128}', name):
self.json(400, {'error': 'Invalid log name'})
return
if path not in ('/api/android-auto/logs', '/api/android-auto/logs/bundle') and name is None:
self.json(404, {'error': 'Not found'})
return
try:
if path == '/api/android-auto/logs':
result = aa_logs().list()
elif name is None:
result = aa_logs().bundle()
else:
result = aa_logs().file(name)
except LogMissing:
self.json(404, {'error': 'That log is no longer kept; refresh the list'})
except LogsUnavailable:
self.json(503, {'error': 'Android Auto logs are unavailable'})
else:
# The session can be revoked while logs are read; check again right before sending them.
if not self.require_session():
return
if path == '/api/android-auto/logs':
self.json(200, result)
else:
filename, body = result
self.attachment(filename, body, 'application/zip' if filename.endswith('.zip') else 'text/plain; charset=utf-8')
elif path.startswith('/api/navigation/map/tiles/'):
if not self.require_session():
return
@@ -50,6 +50,9 @@ assert.match(feed.error, /size limit/)
const packageFile = { size: 32 }
assert.equal(await feed.upload(packageFile), true)
assert.equal(calls.find(([path]) => path.endsWith("/upload"))[1].body, packageFile)
assert.equal(await feed.removePackage(), true)
assert.equal(calls.find(([path]) => path.endsWith("/identity"))[1].method, "DELETE")
assert.match(AndroidAutoPage.template, /Delete Package/)
feed.stop(true)
assert(calls.some(([path, options]) => path.endsWith("/cancel") && options.keepalive))
await feed.start()
@@ -0,0 +1,58 @@
import assert from "node:assert/strict"
import { AndroidAutoLogsFeed, BUNDLE_URL, fileUrl, outcomeLabel } from "../web/js/android-auto-logs.js"
const flush = async () => { for (let i = 0; i < 8; i++) await Promise.resolve() }
function fixture() {
const requests = [], states = []
let unauthorized = 0
const feed = new AndroidAutoLogsFeed({ publish: (state) => states.push(state), unauthorized: () => { unauthorized++ },
fetcher: (url, options) => new Promise((resolve) => requests.push({ url, signal: options.signal, resolve })) })
const reply = async (index, body, status = 200) => {
requests[index].resolve({ ok: status === 200, status, json: async () => body })
await flush()
}
return { feed, requests, states, reply, get unauthorized() { return unauthorized } }
}
assert.equal(BUNDLE_URL, "./api/android-auto/logs/bundle")
assert.equal(fileUrl("session-000002-20261005-094551.jsonl"), "./api/android-auto/logs/file/session-000002-20261005-094551.jsonl")
assert.equal(fileUrl("a b/c"), "./api/android-auto/logs/file/a%20b%2Fc")
assert.equal(outcomeLabel("failed: RFCOMM: timed out"), "Failed: RFCOMM: timed out")
assert.equal(outcomeLabel(""), "Unknown")
const listing = { schemaVersion: 1, sessions: [{ name: "session-000002-x.jsonl", size: 10, modifiedAt: 100, outcome: "projected", car: "Honda Civic" }],
others: [{ name: "car_ui.log", size: 5, modifiedAt: 90 }] }
const normal = fixture()
normal.feed.start()
assert.equal(normal.requests[0].url, "./api/android-auto/logs")
assert.equal(normal.states.at(-1).status, "loading")
await normal.reply(0, listing)
assert.equal(normal.states.at(-1).status, "ready")
assert.deepEqual(normal.states.at(-1).sessions, listing.sessions)
assert.deepEqual(normal.states.at(-1).others, listing.others)
normal.feed.load()
assert.equal(normal.requests[1].url, "./api/android-auto/logs")
normal.feed.stop()
assert.equal(normal.requests[1].signal.aborted, true)
assert.deepEqual(normal.states.at(-1).sessions, [])
const stale = fixture()
stale.feed.start()
stale.feed.load()
await stale.reply(1, listing)
const count = stale.states.length
await stale.reply(0, { schemaVersion: 1, sessions: [], others: [] })
assert.equal(stale.states.length, count)
const revoked = fixture()
revoked.feed.start()
await revoked.reply(0, { error: "Sign in" }, 401)
assert.equal(revoked.unauthorized, 1)
assert.deepEqual(revoked.states.at(-1).sessions, [])
const broken = fixture()
broken.feed.start()
await broken.reply(0, { error: "Android Auto logs are unavailable" }, 503)
assert.equal(broken.states.at(-1).status, "unavailable")
assert.equal(broken.unauthorized, 0)
@@ -0,0 +1,152 @@
"""Android Auto logs in Logs & Diagnostics: the read-only reader and its authenticated download routes."""
import http.client
import io
import json
from pathlib import Path
import tempfile
import threading
import unittest
from unittest import mock
import zipfile
from openpilot.starpilot.galaxy.access import GalaxyAccessOwner
from openpilot.starpilot.galaxy.android_auto_logs import AndroidAutoLogs, LogMissing
from openpilot.starpilot.galaxy.server import make_server
SESSION = "\n".join(json.dumps(event) for event in (
{"t": "2026-10-05T16:45:51.000+00:00", "event": "session_start", "receiver": "Honda CIVIC", "trigger": "auto"},
{"t": "2026-10-05T16:45:52.000+00:00", "event": "stage", "stage": "rfcomm"},
{"t": "2026-10-05T16:46:52.000+00:00", "event": "attempt_failed", "stage": "rfcomm", "error": "RFCOMM: timed out", "kind": "TimeoutError"},
{"t": "2026-10-05T16:46:53.000+00:00", "event": "session_stop"},
)) + "\n"
def fill(directory: Path) -> None:
directory.mkdir(parents=True)
(directory / "session-000001-20261004-221339.jsonl").write_text(SESSION)
(directory / "session-000002-20261005-094551.jsonl").write_text(SESSION)
(directory / "car_ui.log").write_text("car view pipeline: nv12, async readback\n")
(directory / "notes.txt").write_text("not an Android Auto log")
class AndroidAutoLogsReaderTest(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name)
self.logs = self.root / "logs"
fill(self.logs)
self.config = self.root / "config.json"
self.config.write_text(json.dumps({"config_version": 3, "receiver_address": "AA:BB:CC:DD:EE:FF", "receiver_name": "Honda CIVIC"}))
self.reader = AndroidAutoLogs(self.logs, self.config)
def test_lists_sessions_newest_first_with_a_summary_and_only_known_logs(self):
listing = self.reader.list()
self.assertEqual([item["name"] for item in listing["sessions"]],
["session-000002-20261005-094551.jsonl", "session-000001-20261004-221339.jsonl"])
session = listing["sessions"][0]
self.assertTrue(session["outcome"].startswith("failed"))
self.assertEqual(session["trigger"], "auto")
self.assertEqual([item["name"] for item in listing["others"]], ["car_ui.log"])
def test_serves_only_listed_logs(self):
self.assertEqual(self.reader.file("car_ui.log"), ("car_ui.log", b"car view pipeline: nv12, async readback\n"))
for name in ("notes.txt", "../config.json", "missing.jsonl", ""):
with self.assertRaises(LogMissing):
self.reader.file(name)
(self.logs / "render_profile.txt").symlink_to(self.config)
with self.assertRaises(LogMissing):
self.reader.file("render_profile.txt")
self.assertNotIn("render_profile.txt", [item["name"] for item in self.reader.list()["others"]])
def test_bundle_is_a_zip_without_bluetooth_addresses(self):
name, data = self.reader.bundle()
self.assertRegex(name, r"^starpilot-android-auto-logs-\d{8}-\d{6}\.zip$")
with zipfile.ZipFile(io.BytesIO(data)) as archive:
names = archive.namelist()
self.assertIn("logs/session-000002-20261005-094551.jsonl", names)
self.assertIn("reports/session-000002-20261005-094551.json", names)
self.assertIn("logs/car_ui.log", names)
self.assertNotIn("logs/notes.txt", names)
config = archive.read("config.json").decode()
self.assertNotIn("AA:BB:CC:DD:EE:FF", config)
def test_missing_directory_is_an_empty_listing(self):
reader = AndroidAutoLogs(self.root / "absent")
self.assertEqual(reader.list(), {"schemaVersion": 1, "sessions": [], "others": []})
class AndroidAutoLogsHTTPTest(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
root = Path(temporary.name)
fill(root / "logs")
(root / "config.json").write_text("{}")
self.owner = GalaxyAccessOwner(root / "access")
self.reader = AndroidAutoLogs(root / "logs", root / "config.json")
self.server = make_server(port=0, owner=self.owner, android_auto_logs=self.reader)
self.thread = threading.Thread(target=self.server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
self.thread.start()
self.addCleanup(self.close_server)
def close_server(self):
self.server.shutdown()
self.thread.join(timeout=2)
self.server.server_close()
def request(self, path, *, method="GET", body=None, headers=None):
connection = http.client.HTTPConnection("127.0.0.1", self.server.server_port, timeout=2)
try:
connection.request(method, path, body=body, headers={"Forwarded": "for=203.0.113.8", **(headers or {})})
response = connection.getresponse()
return response.status, response.read(), dict(response.getheaders())
finally:
connection.close()
def login(self):
status, _, headers = self.request("/api/auth/login", method="POST", body=json.dumps({"password": "password123"}),
headers={"Content-Type": "application/json", "Origin": f"http://127.0.0.1:{self.server.server_port}"})
self.assertEqual(status, 200)
return headers["Set-Cookie"].split(";", 1)[0]
def test_requires_a_session_before_reading_anything(self):
self.assertTrue(self.owner.configure("password123", lambda: True))
with mock.patch.object(self.reader, "list", wraps=self.reader.list) as listing, \
mock.patch.object(self.reader, "bundle", wraps=self.reader.bundle) as bundle:
for path in ("/api/android-auto/logs", "/api/android-auto/logs/bundle", "/api/android-auto/logs/file/car_ui.log"):
self.assertEqual(self.request(path)[0], 401)
listing.assert_not_called()
bundle.assert_not_called()
def test_list_file_and_bundle_downloads(self):
self.assertTrue(self.owner.configure("password123", lambda: True))
cookie = {"Cookie": self.login()}
status, body, headers = self.request("/api/android-auto/logs", headers=cookie)
self.assertEqual(status, 200)
self.assertEqual(headers["Cache-Control"], "no-store")
self.assertEqual(len(json.loads(body)["sessions"]), 2)
status, body, headers = self.request("/api/android-auto/logs/file/session-000001-20261004-221339.jsonl", headers=cookie)
self.assertEqual(status, 200)
self.assertEqual(headers["Content-Disposition"], 'attachment; filename="session-000001-20261004-221339.jsonl"')
self.assertEqual(body.decode(), SESSION)
status, body, headers = self.request("/api/android-auto/logs/bundle", headers=cookie)
self.assertEqual(status, 200)
self.assertEqual(headers["Content-Type"], "application/zip")
self.assertRegex(headers["Content-Disposition"], r'^attachment; filename="starpilot-android-auto-logs-\d{8}-\d{6}\.zip"$')
with zipfile.ZipFile(io.BytesIO(body)) as archive:
self.assertIn("REPORT.txt", archive.namelist())
def test_unknown_and_invalid_names(self):
self.assertTrue(self.owner.configure("password123", lambda: True))
cookie = {"Cookie": self.login()}
self.assertEqual(self.request("/api/android-auto/logs/file/notes.txt", headers=cookie)[0], 404)
self.assertEqual(self.request("/api/android-auto/logs/file/..%2Fconfig.json", headers=cookie)[0], 400)
self.assertEqual(self.request("/api/android-auto/logs/other", headers=cookie)[0], 404)
if __name__ == "__main__":
unittest.main()
@@ -961,6 +961,10 @@ button.gx-crash-row:hover { background: var(--glass-bg-hover); transform: transl
.gx-crash-row span { flex: 0 0 auto; color: var(--text-muted); font-size: var(--fs-sm); }
.gx-crash-preview { margin-top: var(--sp-3); }
.gx-crash-preview__head { display: flex; align-items: center; justify-content: space-between; gap: var(--sp-2); overflow-wrap: anywhere; }
.gx-aa-log { cursor: default; }
.gx-aa-log > div { display: grid; gap: 2px; min-width: 0; }
.gx-aa-log > div span { overflow-wrap: anywhere; }
.gx-aa-log > a { flex: 0 0 auto; }
.gx-crash-preview pre { white-space: pre-wrap; overflow-wrap: anywhere; max-height: 60vh; overflow-y: auto; font-size: var(--fs-sm); }
.gx-software-card { margin-top: var(--sp-3); padding: var(--sp-4); }
.gx-software-card h3 { margin: 0 0 var(--sp-3); }
@@ -0,0 +1,70 @@
// Read-only Android Auto session logs. State is cleared on navigation and sign-out.
export const LOGS_URL = "./api/android-auto/logs"
export const BUNDLE_URL = `${LOGS_URL}/bundle`
export const fileUrl = (name) => `${LOGS_URL}/file/${encodeURIComponent(name)}`
// compat_report's outcome is a short phrase: "projected", "failed: <error>", "stopped at rfcomm", "no session".
export const outcomeLabel = (outcome) => (outcome ? outcome[0].toUpperCase() + outcome.slice(1) : "Unknown")
export class AndroidAutoLogsFeed {
constructor({ publish, unauthorized = () => {}, fetcher = (...args) => fetch(...args) }) {
Object.assign(this, { publish, unauthorized, fetcher })
this.active = false
this.generation = 0
this.request = null
}
clear() {
this.publish({ sessions: [], others: [], status: "idle", error: "" })
}
stop() {
this.active = false
this.generation++
this.request?.abort()
this.request = null
this.clear()
}
start() {
this.stop()
this.active = true
return this.load()
}
async load() {
if (!this.active) return
const generation = ++this.generation
this.request?.abort()
const request = new AbortController()
this.request = request
const current = () => this.active && generation === this.generation && !request.signal.aborted
this.publish({ status: "loading", error: "" })
try {
const response = await this.fetcher(LOGS_URL, { signal: request.signal, cache: "no-store" })
if (!current()) return
if (response.status === 401) {
this.stop()
this.unauthorized()
return
}
if (response.status === 503) {
const body = await response.json().catch(() => null)
if (!current()) return
if (["access_unavailable", "setup_required"].includes(body?.code)) {
this.stop()
this.unauthorized()
return
}
throw new Error("Android Auto logs are unavailable")
}
if (!response.ok) throw new Error("Android Auto logs are unavailable")
const body = await response.json()
if (!current()) return
this.publish({ sessions: body.sessions ?? [], others: body.others ?? [], status: "ready", error: "" })
} catch (error) {
if (!current() || error?.name === "AbortError") return
this.publish({ sessions: [], others: [], status: "unavailable", error: error?.message || "Android Auto logs are unavailable" })
}
}
}
@@ -279,6 +279,15 @@ export class AndroidAutoFeed {
await this.refresh()
return true
}
async removePackage() {
if (!this.active || this.busy || !this.setup?.parked || !this.setup?.enabled ||
this.setup.import?.state === "running" || this.runtime?.running) return false
const result = await this.request("./api/android-auto/identity", { method: "DELETE" })
if (result === null) return false
await this.refresh()
return true
}
}
export const AndroidAutoPage = {
@@ -316,6 +325,14 @@ export const AndroidAutoPage = {
if (this.busy) return "Waiting for Android Auto setup to respond…"
return ""
},
hasPackage() { return !!(this.setup?.identity.installed || this.setup?.identity.expired || this.setup?.identity.error) },
removeReason() {
if (!this.setup?.parked) return "Use offroad mode or Park to delete the package."
if (!this.setup.enabled) return "Enable Android Auto to delete the package."
if (this.setup.import?.state === "running") return "Wait for the current package verification to finish."
if (this.runtime?.running) return "Stop projection before deleting the package."
return ""
},
pairingReason() {
if (!this.setup) return "Load setup status first."
if (!this.setup.installReady) return "This build is missing the Android Auto display or encoder."
@@ -354,6 +371,11 @@ export const AndroidAutoPage = {
if (!this.setup.enabled && !await this.feed.setEnabled(true)) return false
return this.feed.upload(file)
},
removePackage() {
if (this.removeReason) return false
if (!window.confirm("Delete the Android Auto package and its certificate from this device? Projection will not start until you upload a package again.")) return false
return this.feed.removePackage()
},
loadReceivers() { return this.feed.loadReceivers() },
control(action, extra = {}) { return this.feed.control(action, extra) },
},
@@ -385,6 +407,8 @@ export const AndroidAutoPage = {
<p v-if="setup.import?.state === 'running'" role="status">Checking your package…</p>
<GxNotice tone="danger" v-if="setup.import?.state === 'failed'">Package verification failed: {{ setup.import.error || 'Try another package.' }}</GxNotice>
<p v-if="setup.import?.state === 'done' && setup.identity.installed" role="status">Package verified and ready.</p>
<div v-if="hasPackage" class="gx-driving__actions"><button class="gx-btn gx-btn--tonal" :disabled="busy || !!removeReason" :title="removeReason" @click="removePackage">Delete Package</button></div>
<p v-if="hasPackage && removeReason && setup.import?.state !== 'running'" class="gx-muted">{{ removeReason }}</p>
<div class="gx-driving__actions"><input class="gx-field" type="file" accept=".apk,.xapk,.apkm" aria-label="Android Auto APK, XAPK, or APKM" @change="choosePackage" />
<GxIconButton v-if="setup.enabled" label="Upload package" icon="bi-upload" :disabled="!!uploadReason" @click="upload" />
<button v-else class="gx-btn" :disabled="!!uploadReason" @click="upload">Enable &amp; upload</button></div>
+41 -4
View File
@@ -3,25 +3,33 @@ import { TroubleshootPage, TmuxPage } from "./diagnostic-tools.js"
import { navigate } from "./router.js"
import { SystemMonitor } from "./system-monitor.js"
import { CrashReportsFeed } from "./crash-reports.js"
import { AndroidAutoLogsFeed, BUNDLE_URL, fileUrl, outcomeLabel } from "./android-auto-logs.js"
import { MenuTile } from "./menu-tile.js"
export const Logs = {
name: "Logs",
components: { GxNotice, SystemMonitor, TroubleshootPage, TmuxPage, MenuTile },
props: { path: { type: String, required: true }, mode: { type: String, required: true }, unauthorized: { type: Function, required: true } },
data: () => ({ crashes: { reports: [], scanIncomplete: false, listLimited: false, status: "idle", error: "", selected: null, preview: null, previewStatus: "idle" }, search: "" }),
data: () => ({ crashes: { reports: [], scanIncomplete: false, listLimited: false, status: "idle", error: "", selected: null, preview: null, previewStatus: "idle" }, search: "",
aaLogs: { sessions: [], others: [], status: "idle", error: "" } }),
computed: {
visibleReports() { return this.crashes.reports.filter((item) => item.name.toLowerCase().includes(this.search.toLowerCase())) },
},
created() {
this.crashFeed = new CrashReportsFeed({ publish: (update) => Object.assign(this.crashes, update), unauthorized: this.unauthorized })
this.aaFeed = new AndroidAutoLogsFeed({ publish: (update) => Object.assign(this.aaLogs, update), unauthorized: this.unauthorized })
},
mounted() { if (this.path === "/logs/crashes" && this.mode === "local") this.crashFeed.start() },
beforeUnmount() { this.crashFeed.stop() },
mounted() {
if (this.path === "/logs/crashes" && this.mode === "local") this.crashFeed.start()
if (this.path === "/logs/android-auto" && this.mode === "local") this.aaFeed.start()
},
beforeUnmount() { this.crashFeed.stop(); this.aaFeed.stop() },
watch: {
path(next, previous) {
if (next === "/logs/crashes" && this.mode === "local") this.crashFeed.start()
else if (previous === "/logs/crashes") { this.crashFeed.stop(); this.search = "" }
if (next === "/logs/android-auto" && this.mode === "local") this.aaFeed.start()
else if (previous === "/logs/android-auto") this.aaFeed.stop()
},
},
methods: {
@@ -30,6 +38,10 @@ export const Logs = {
openTmux() { navigate("/logs/tmux") },
openMonitor() { navigate("/logs/monitor") },
openCrashes() { if (this.mode === "local") navigate("/logs/crashes") },
openAndroidAuto() { if (this.mode === "local") navigate("/logs/android-auto") },
bundleUrl: () => BUNDLE_URL,
fileUrl,
outcomeLabel,
reportDate(seconds) { return new Date(seconds * 1000).toLocaleString() },
reportSize(bytes) { return `${(bytes / 1024).toFixed(1)} KiB` },
async copyPreview() {
@@ -41,11 +53,13 @@ export const Logs = {
template: `
<div class="gx-view">
<h2 v-if="path === '/logs'">Logs & Diagnostics</h2>
<p v-if="path === '/logs'" class="gx-note">Live console, driving configuration, system activity and crash reports.</p>
<p v-if="path === '/logs'" class="gx-note">Live console, driving configuration, system activity, crash reports and Android Auto logs.</p>
<div v-if="path === '/logs'" class="gx-grid">
<MenuTile icon="bi-cpu" title="System Monitor" :description="mode === 'sample' ? 'Inspect a synthetic activity sample' : 'Inspect local system activity'" @select="openMonitor" />
<MenuTile icon="bi-file-text" title="Crash Reports" :description="mode === 'local' ? 'Read local reports' : ''"
:availability="mode === 'local' ? '' : 'Unavailable in preview'" :disabled="mode !== 'local'" @select="openCrashes" />
<MenuTile icon="bi-phone" title="Android Auto" :description="mode === 'local' ? 'Session logs and a download for bug reports' : ''"
:availability="mode === 'local' ? '' : 'Unavailable in preview'" :disabled="mode !== 'local'" @select="openAndroidAuto" />
<MenuTile icon="bi-terminal" title="tmux Live View" description="Read the launcher console tail" @select="openTmux" />
<MenuTile icon="bi-wrench" title="Troubleshoot" description="Read device and vehicle diagnostics" @select="openTroubleshoot" />
</div>
@@ -68,6 +82,29 @@ export const Logs = {
<section v-else-if="crashes.preview" class="gx-card gx-crash-preview"><div class="gx-crash-preview__head"><strong>{{ crashes.preview.name }}</strong><button type="button" class="gx-btn gx-btn--tonal" @click="copyPreview">Copy visible text</button></div><p v-if="crashes.preview.truncated" class="gx-note">Preview truncated to the first 256 KiB.</p><pre>{{ crashes.preview.text }}</pre></section>
<GxNotice tone="danger" v-if="crashes.error">{{ crashes.error }}</GxNotice>
</template>
<template v-else-if="path === '/logs/android-auto' && mode === 'local'">
<h2>Android Auto Logs</h2>
<p class="gx-note">The last 20 connection sessions, newest first. The download is one zip with every session, a readable report for each, the Android Auto settings without Bluetooth addresses, and the car view's renderer logs.</p>
<div class="gx-crash-controls"><a class="gx-btn" :href="bundleUrl()" download><i class="bi bi-download"></i> Download all (zip)</a><button type="button" class="gx-btn gx-btn--tonal" @click="aaFeed.load()">Refresh</button></div>
<p v-if="aaLogs.status === 'loading'">Loading Android Auto logs…</p>
<p v-else-if="aaLogs.status === 'unavailable'" role="alert">{{ aaLogs.error || 'Android Auto logs are unavailable.' }}</p>
<p v-else-if="aaLogs.status === 'ready' && !aaLogs.sessions.length">No Android Auto sessions have been logged yet.</p>
<div v-if="aaLogs.status === 'ready'" class="gx-crash-list">
<div v-for="log in aaLogs.sessions" :key="log.name" class="gx-card gx-crash-row gx-aa-log">
<div><strong>{{ outcomeLabel(log.outcome) }}</strong><span>{{ log.car || 'Car not identified' }}{{ log.transport ? ' · ' + log.transport : '' }}{{ log.trigger ? ' · ' + log.trigger : '' }} · {{ reportDate(log.modifiedAt) }} · {{ reportSize(log.size) }}</span></div>
<a class="gx-btn gx-btn--tonal" :href="fileUrl(log.name)" download :aria-label="'Download ' + log.name"><i class="bi bi-download"></i></a>
</div>
</div>
<template v-if="aaLogs.status === 'ready' && aaLogs.others.length">
<h3>Renderer logs</h3>
<div class="gx-crash-list">
<div v-for="log in aaLogs.others" :key="log.name" class="gx-card gx-crash-row gx-aa-log">
<div><strong>{{ log.name }}</strong><span>{{ reportDate(log.modifiedAt) }} · {{ reportSize(log.size) }}</span></div>
<a class="gx-btn gx-btn--tonal" :href="fileUrl(log.name)" download :aria-label="'Download ' + log.name"><i class="bi bi-download"></i></a>
</div>
</div>
</template>
</template>
<div v-else class="gx-card gx-message" role="status">This Logs & Diagnostics page is unavailable in the offline preview.</div>
</div>`,
}
@@ -214,7 +214,8 @@ class BluezPhone:
# AutoConnect stays off: BlueZ would otherwise open this link to every
# device with a hands-free role (headsets, other gadgets) when it connects.
options = {"Name": ("s", "StarPilot Hands-Free Gateway"), "RequireAuthentication": ("b", True),
"RequireAuthorization": ("b", False), "AutoConnect": ("b", False)}
"RequireAuthorization": ("b", False), "AutoConnect": ("b", False),
"Version": ("q", hfp.HFP_VERSION), "Features": ("q", hfp.SDP_FEATURES)}
try:
self._call("/org/bluez", PROFILE_MANAGER_IFACE, "RegisterProfile", "osa{sv}", (HFP_PROFILE_PATH, HFP_AG_UUID, options))
self._profile_registered = True
@@ -2,7 +2,9 @@
The head unit is the RFCOMM server; StarPilot connects as the phone and runs:
HU -> WifiStartRequest(ip, port) [some receivers first send WifiVersionRequest]
HU -> WifiStartRequest(ip, port) [some receivers first send WifiVersionRequest; others wait
for the phone's WifiStartRequest and answer with
WifiStartResponse(ip, port, status)]
ph -> WifiInfoRequest()
HU -> WifiInfoResponse(ssid, key, bssid, security, ap_type)
ph -> WifiStartResponse(status=0)
@@ -31,7 +33,7 @@ import time
from collections.abc import Callable
from dataclasses import dataclass, field as dataclass_field
from openpilot.starpilot.system.android_auto.wire import field, one, parse_fields, signed, text
from openpilot.starpilot.system.android_auto.wire import describe, field, one, parse_fields, signed, text
WIFI_START_REQUEST = 1
WIFI_INFO_REQUEST = 2
@@ -58,6 +60,7 @@ SECURITY_NAMES = {0: "unknown", 1: "open", 2: "wep64", 3: "wep128", 4: "wpa", 8:
MAX_FRAME = 4096
MAX_FRAMES = 64
INITIAL_KICK_SECONDS = 2.5
MAC_RE = re.compile(r"^(?:[0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$")
@@ -67,6 +70,10 @@ class BootstrapError(RuntimeError):
self.stage = stage
class BootstrapTimeout(BootstrapError):
pass
@dataclass(frozen=True)
class Endpoint:
ip: str
@@ -198,12 +205,25 @@ class FrameReader:
return frames
def describe_version_request(payload: bytes):
"""The whole WifiVersionRequest for diagnostics, without the vehicle identifier some receivers include."""
described = describe(payload)
if isinstance(described, dict):
for number in (4, 5):
for info in described.get(number, []):
if isinstance(info, dict) and isinstance((info.get(4) or [None])[0], str):
info[4] = ["redacted"] # HeadUnitInfo.vehicle_id; the endpoint's field 4 is a number
return described
class WirelessBootstrap:
"""Runs the phone side of the handshake on a connected RFCOMM socket."""
def __init__(self, sock, log: Callable[..., None], *, device_serial: str = "starpilot",
version_status: int = STATUS_SUCCESS, stage_timeout: float = 20.0, start_request_delay: float = 5.0):
version_status: int = STATUS_SUCCESS, stage_timeout: float = 20.0, start_request_delay: float = 5.0,
initial_kick_delay: float = INITIAL_KICK_SECONDS):
self.sock = sock
self.initial_kick_delay = initial_kick_delay
self.start_request_delay = start_request_delay
self.log = log
self.device_serial = device_serial
@@ -227,7 +247,7 @@ class WirelessBootstrap:
while not self.queue:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise BootstrapError(self.stage, "head unit did not answer in time")
raise BootstrapTimeout(self.stage, "head unit did not answer in time")
if self.cancelled():
raise BootstrapError(self.stage, "cancelled")
# Short slices keep Stop responsive even where closing a socket from
@@ -274,42 +294,69 @@ class WirelessBootstrap:
# Android Auto 17.6 asks the car to start projection 5 s after the version
# exchange if the car has not started it; newer head units (2025 Honda) wait for it.
start_request_at: float | None = None
deadline = time.monotonic() + self.stage_timeout
initial_kick_at: float | None = time.monotonic() + self.initial_kick_delay
hint_ready_at: float | None = None
while endpoint is None:
wait = 3.0 if hinted is not None else self.stage_timeout
if start_request_at is not None:
wait = min(wait, max(0.01, start_request_at - time.monotonic()))
try:
message_id, payload = self.next_frame(wait)
except BootstrapError:
if start_request_at is not None and time.monotonic() >= start_request_at:
start_request_at = None
self.send(WIFI_START_REQUEST)
continue
if hinted is None:
raise
endpoint = hinted # the receiver sent its endpoint only with the version exchange
if cancelled():
raise BootstrapError(self.stage, "cancelled")
now = time.monotonic()
# An endpoint hint gets a short grace period for the normal StartRequest.
# Pings must not keep extending that grace period (or either stage).
if hinted is not None and hint_ready_at is not None and now >= hint_ready_at:
endpoint = hinted
break
if now >= deadline:
raise BootstrapTimeout(self.stage, "head unit did not answer in time")
if initial_kick_at is not None and now >= initial_kick_at:
initial_kick_at = None
self.log("bootstrap_initial_kick")
self.send(WIFI_START_REQUEST)
if start_request_at is not None and now >= start_request_at:
start_request_at = None
self.send(WIFI_START_REQUEST)
wake_at = min(t for t in (deadline, initial_kick_at, start_request_at, hint_ready_at) if t is not None)
try:
message_id, payload = self.next_frame(max(0.001, wake_at - time.monotonic()))
except BootstrapTimeout:
continue # only an actual read timeout advances timers; EOF/cancellation must fail
if cancelled():
raise BootstrapError(self.stage, "cancelled")
if self.service(message_id, payload):
continue
initial_kick_at = None # setup messages drive the exchange; pings alone do not
if message_id == WIFI_VERSION_REQUEST:
major, minor, version_endpoint, info = parse_version_request(payload)
version, head_unit = (major, minor), {**head_unit, **info}
self.log("bootstrap_version", major=major, minor=minor, head_unit=info,
endpoint=version_endpoint.__dict__ if version_endpoint else None)
endpoint=version_endpoint.__dict__ if version_endpoint else None, message=describe_version_request(payload))
self.send(WIFI_VERSION_RESPONSE, field(1, major) + field(2, minor) + field(3, self.device_serial) +
field(4, self.version_status))
hinted = version_endpoint or hinted
start_request_at = time.monotonic() + self.start_request_delay
if hinted is not None and hint_ready_at is None:
hint_ready_at = time.monotonic() + 3.0
if start_request_at is None:
start_request_at = time.monotonic() + self.start_request_delay
elif message_id == WIFI_START_REQUEST:
parsed = parse_endpoint(payload)
if parsed is None:
raise BootstrapError(self.stage, "WifiStartRequest has no valid IPv4 endpoint")
endpoint = parsed
elif message_id == WIFI_START_RESPONSE:
# The car's answer to our WifiStartRequest (2025 Honda): ip=1, port=2, status=3 once it is
# ready, or a negative status alone before then (it may follow up on its own; ask again anyway).
status = signed(one(parse_fields(payload), 3, STATUS_SUCCESS))
parsed = parse_endpoint(payload)
if parsed is not None and status == STATUS_SUCCESS:
endpoint = parsed
else:
self.log("bootstrap_start_refused", status=status, endpoint=parsed is not None)
start_request_at = time.monotonic() + self.start_request_delay
elif message_id == WIFI_SETUP_INFO:
setup_endpoint, setup_credentials = parse_setup_info(payload)
hinted = setup_endpoint or hinted
if hinted is not None and hint_ready_at is None:
hint_ready_at = time.monotonic() + 3.0
credentials = setup_credentials or credentials
self.log("bootstrap_setup_info", endpoint=setup_endpoint.__dict__ if setup_endpoint else None,
credentials=credentials.describe() if credentials else None)
@@ -322,8 +369,12 @@ class WirelessBootstrap:
if credentials is None:
self.stage = "wifi_info"
self.send(WIFI_INFO_REQUEST)
info_deadline = time.monotonic() + self.stage_timeout
while credentials is None:
message_id, payload = self.next_frame(self.stage_timeout)
remaining = info_deadline - time.monotonic()
if remaining <= 0:
raise BootstrapTimeout(self.stage, "head unit did not answer in time")
message_id, payload = self.next_frame(remaining)
if cancelled():
raise BootstrapError(self.stage, "cancelled")
if self.service(message_id, payload):
@@ -0,0 +1,304 @@
"""A short, shareable account of how one Android Auto session went with one car.
Reads a session log (``/data/android_auto/logs/session-*.jsonl``) and reports what a
compatibility report needs: which car and head unit, wired or wireless, how far the
connection got and why it stopped, the Wi-Fi security, TLS suite and video modes the
car offered, and the messages StarPilot does not act on yet. Used by
``openpilot/tools/android_auto/compat_report.py``.
The report carries no Wi-Fi name, key, address or vehicle identifier; the raw logs,
which the bundle also includes, already leave out keys and vehicle ids.
"""
from __future__ import annotations
import io
import json
import os
import stat
import re
import zipfile
from pathlib import Path
from openpilot.starpilot.system.android_auto import identity as identity_store
from openpilot.starpilot.system.android_auto.session import describe_video_config
SESSION_GLOB = "session-*.jsonl"
SESSION_NAME = re.compile(r"session-(?:\d{6}-)?[\w.-]+\.jsonl")
MAX_LOG_BYTES = 8 << 20
MAX_BUNDLE_BYTES = 64 << 20
MAX_SESSION_LOGS = 20
MAX_EVENTS = 200_000 # a day of 30 s stats is ~3000 lines; this only stops a runaway file
EXTRA_LOGS = ("render_profile.txt", "render_profile.1.txt", "car_ui.log")
EXTRA_LOG_BYTES = 1 << 20 # the tail of each extra log kept in a bundle
# Connection stages in the order a session passes them, for "how far did it get".
STAGE_ORDER = ("connecting_bluetooth", "discovering", "rfcomm", "wifi_start", "wifi_info", "joining_wifi", "connecting_tcp",
"waiting_for_usb", "usb_accessory", "authenticating", "negotiating", "streaming")
# ServiceDiscoveryResponse and its headunit_info, by field number (JSON turns the numbers into strings).
SDR_FIELDS = {"2": "car_make", "3": "car_model", "4": "car_year", "7": "head_unit_make", "8": "head_unit_model",
"9": "head_unit_software_build", "10": "head_unit_software_version", "14": "display_name"}
HEAD_UNIT_INFO_FIELDS = {"1": "car_make", "2": "car_model", "3": "car_year", "5": "head_unit_make", "6": "head_unit_model",
"7": "head_unit_software_build", "8": "head_unit_software_version"}
IGNORED_EVENTS = ("control_ignored", "channel_ignored", "video_ignored", "input_ignored", "unexpected_while_waiting",
"handshake_ignored", "bootstrap_ignored")
def read_log(path: Path, limit: int, *, tail: bool = False) -> bytes:
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
try:
fd = os.open(path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
finally:
os.close(directory)
with os.fdopen(fd, "rb") as handle:
info = os.fstat(handle.fileno())
if not stat.S_ISREG(info.st_mode):
raise OSError("Not a regular Android Auto log")
if tail:
handle.seek(max(0, info.st_size - limit))
elif info.st_size > limit:
raise OSError("Android Auto log exceeds the size limit")
data = handle.read(limit + 1)
if len(data) > limit:
raise OSError("Android Auto log exceeds the size limit")
return data
def events_from_bytes(data: bytes) -> list[dict]:
events = []
for line in data.decode("utf-8", "replace").splitlines():
if len(events) >= MAX_EVENTS:
break
try:
record = json.loads(line)
except ValueError:
continue
if isinstance(record, dict) and isinstance(record.get("event"), str):
events.append(record)
return events
def load_events(path: Path) -> list[dict]:
try:
return events_from_bytes(read_log(path, MAX_LOG_BYTES))
except OSError:
return []
def _first_text(values) -> str:
if isinstance(values, list) and values and isinstance(values[0], str):
return values[0]
return ""
def _car_from_discovery(head_unit: dict) -> dict:
car = {label: _first_text(head_unit.get(number)) for number, label in SDR_FIELDS.items()}
info = head_unit.get("17")
if isinstance(info, list) and info and isinstance(info[0], dict):
for number, label in HEAD_UNIT_INFO_FIELDS.items():
car[label] = _first_text(info[0].get(number)) or car.get(label, "")
return {key: value for key, value in car.items() if value}
def _video_offer(channels: list | None) -> list[str]:
"""The logged video configurations (JSON turned their field numbers into strings), named as the session names them."""
offered = []
for channel in channels if isinstance(channels, list) else []:
if not isinstance(channel, dict):
continue
for config in channel.get("video_configs") or []:
if isinstance(config, dict):
fields = {int(number): values for number, values in config.items() if str(number).isdigit() and isinstance(values, list)}
offered.append(describe_video_config(fields, channel.get("display_type")))
return offered
def summarize(events: list[dict]) -> dict:
"""Everything a compatibility report needs from one session log, as plain JSON-able values."""
report: dict = {"started": events[0].get("t", "") if events else "", "events": len(events), "transport": "",
"trigger": "", "car": {}, "stages": [], "furthest_stage": "", "outcome": "no session",
"errors": [], "ended": [], "wifi": {}, "tls": {}, "video": {}, "usb": {}, "bluetooth": {},
"focus": {"granted": 0, "lost": 0}, "ignored": [], "stats": {}}
ignored: dict[tuple, dict] = {}
projected = False # the car acknowledged a frame; focus or the "streaming" stage alone do not show that
failed_after = None
for record in events:
name = record["event"]
if name == "session_start":
report["transport"] = "wired" if record.get("receiver") == "usb" else "wireless"
report["trigger"] = record.get("trigger", "")
if record.get("receiver") and record.get("receiver") != "usb":
report["car"].setdefault("bluetooth_name", record["receiver"])
elif name == "stage":
state = record.get("state", "")
if state and (not report["stages"] or report["stages"][-1] != state):
report["stages"].append(state)
elif name == "bootstrap_version" and isinstance(record.get("head_unit"), dict):
report["car"].update({key: value for key, value in record["head_unit"].items() if isinstance(value, str) and value})
report["wifi"]["version"] = f"{record.get('major')}.{record.get('minor')}"
elif name == "discovered":
report["car"].update(_car_from_discovery(record.get("head_unit") or {}))
report["video"]["offered"] = _video_offer(record.get("channels"))
services = sorted({service for channel in record.get("channels") or [] if isinstance(channel, dict)
for service in channel.get("services") or []})
report["video"]["car_services"] = services
elif name == "bootstrap_credentials":
report["wifi"].update(security=record.get("security", ""), ap_type=record.get("ap_type"))
elif name == "wifi_joined":
report["wifi"]["joined"] = True
elif name == "version":
report["tls"]["protocol"] = f"{record.get('major')}.{record.get('minor')} (replied {record.get('reply')})"
elif name == "tls_established":
report["tls"].update(version=record.get("version", ""), cipher=record.get("cipher", ""))
elif name == "tls_failed":
report["tls"]["failed"] = record.get("reason") or record.get("error", "")
elif name == "head_unit_verified":
report["tls"]["head_unit"] = record.get("subject", "")
elif name == "authentication_rejected":
report["tls"]["rejected_status"] = record.get("status")
elif name == "projection_ready":
mode = record.get("mode") or {}
margins = f"{mode.get('margin_width')}x{mode.get('margin_height')}"
report["video"]["chosen"] = f"{mode.get('width')}x{mode.get('height')} @ {mode.get('fps')} fps, margins {margins}"
elif name == "video_focus":
report["focus"]["granted" if record.get("focused") else "lost"] += 1
elif name == "video_acknowledged":
projected = True
failed_after = None
elif name == "usb_gadget_prepared":
report["usb"]["started_as"] = record.get("mode", "")
elif name == "usb_no_accessory_start":
report["usb"]["no_handshake_after_s"] = record.get("waited")
elif name == "usb_accessory_ready":
report["usb"]["method"] = record.get("method", "")
strings = record.get("strings") or {}
report["usb"]["accessory_strings"] = strings # AOA requires "Android" / "Android Auto"; description and version vary
elif name == "usb_state":
report["usb"].setdefault("states", [])
if not report["usb"]["states"] or report["usb"]["states"][-1] != record.get("state"):
report["usb"]["states"] = (report["usb"]["states"] + [record.get("state")])[-12:]
elif name == "hfp_wait":
report["bluetooth"]["hands_free_before_rfcomm"] = record.get("linked")
elif name == "hfp_connected":
report["bluetooth"]["hands_free"] = True
elif name == "rfcomm_channel":
report["bluetooth"]["rfcomm_channel"] = record.get("channel")
elif name == "attempt_failed":
error = {"stage": record.get("stage", ""), "error": record.get("error", "")}
if error not in report["errors"]:
report["errors"].append(error)
if projected:
failed_after = error["error"]
elif name == "session_ended" and record.get("reason"):
report["ended"].append(record["reason"])
elif name == "stats":
report["stats"] = {key: value for key, value in record.items() if key not in ("t", "event")}
elif name in IGNORED_EVENTS:
key = (name, record.get("channel"), record.get("kind", record.get("message")) if name == "bootstrap_ignored" else record.get("kind"))
entry = ignored.setdefault(key, {"event": name, "channel": record.get("channel"), "kind": key[2], "count": 0})
entry["count"] = max(entry["count"] + 1, int(record.get("count") or 0))
if "message" in record and name != "bootstrap_ignored":
entry["example"] = record["message"]
report["ignored"] = sorted(ignored.values(), key=lambda entry: -entry["count"])[:20]
reached = [stage for stage in report["stages"] if stage in STAGE_ORDER]
report["furthest_stage"] = max(reached, key=STAGE_ORDER.index) if reached else ""
if projected:
report["outcome"] = f"projected, then failed: {failed_after}" if failed_after else "projected"
elif report["errors"]:
report["outcome"] = f"failed: {report['errors'][-1]['error']}"
elif events:
report["outcome"] = f"stopped at {report['furthest_stage'] or 'start'}"
return report
def render_text(report: dict, name: str = "") -> str:
"""The report as plain text, to paste into an issue or a message."""
lines = [f"StarPilot Android Auto compatibility report{f' ({name})' if name else ''}",
f"Started: {report['started'] or 'unknown'} Connection: {report['transport'] or 'unknown'} Trigger: {report['trigger'] or '-'}",
f"Outcome: {report['outcome']}"]
car = report["car"]
if car:
lines.append("Car: " + ", ".join(f"{key.replace('_', ' ')}: {value}" for key, value in car.items()))
lines.append("Stages: " + (" > ".join(report["stages"]) or "none"))
for error in report["errors"][-5:]:
lines.append(f"Error ({error['stage']}): {error['error']}")
for reason in report["ended"][-3:]:
lines.append(f"Ended by car: {reason}")
for title, section in (("Wi-Fi", report["wifi"]), ("Bluetooth", report["bluetooth"]), ("TLS", report["tls"]),
("USB", report["usb"]), ("Video", report["video"])):
if section:
lines.append(f"{title}: " + "; ".join(f"{key.replace('_', ' ')}: {value}" for key, value in section.items()))
lines.append(f"Focus: granted {report['focus']['granted']}, lost {report['focus']['lost']}")
if report["stats"]:
lines.append("Last stats: " + json.dumps(report["stats"], default=str)[:400])
for entry in report["ignored"][:10]:
lines.append(f"Not handled: {entry['event']} channel {entry['channel']} kind {entry['kind']} x{entry['count']}"
+ (f" e.g. {json.dumps(entry['example'], default=str)[:200]}" if entry.get("example") is not None else ""))
return "\n".join(lines) + "\n"
def session_logs(log_dir: Path | None = None) -> list[Path]:
"""Session logs, newest first."""
directory = log_dir or identity_store.LOG_DIR
try:
logs = [path for path in directory.glob(SESSION_GLOB) if not path.is_symlink() and path.is_file() and SESSION_NAME.fullmatch(path.name)]
except OSError:
return []
return sorted(logs, key=identity_store.session_log_order, reverse=True)[:MAX_SESSION_LOGS]
def session_path(name: str, log_dir: Path | None = None) -> Path | None:
"""The log called ``name``, only if it is one of the session logs (no paths from outside)."""
return next((path for path in session_logs(log_dir) if path.name == name), None)
def _tail(path: Path, limit: int) -> bytes | None:
try:
return read_log(path, limit, tail=True)
except OSError:
return None
def shareable_config(config: dict) -> dict:
"""The settings without the Bluetooth addresses of the car and its companion device."""
shared = dict(config)
for key in ("receiver_address", "companion_address"):
if shared.get(key):
shared[key] = "redacted"
shared["rfcomm_cache"] = sorted((shared.get("rfcomm_cache") or {}).values()) # the channels, not whose they are
return shared
def bundle(log_dir: Path | None = None, config_path: Path | None = None) -> bytes:
"""A zip for a bug report: every session log with its report, the settings and the renderer logs; never the identity
or a Bluetooth address."""
directory = log_dir or identity_store.LOG_DIR
logs = session_logs(directory)
output = io.BytesIO()
total = 0
with zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as archive:
def add(name: str, data: bytes):
nonlocal total
total += len(data)
if total > MAX_BUNDLE_BYTES:
raise OSError("Android Auto bundle exceeds the size limit")
archive.writestr(name, data)
summaries = []
for path in logs:
try:
data = read_log(path, MAX_LOG_BYTES)
except FileNotFoundError:
continue
report = summarize(events_from_bytes(data))
summaries.append(render_text(report, path.name))
add(f"logs/{path.name}", data)
add(f"reports/{path.stem}.json", json.dumps(report, indent=2, default=str).encode())
add("REPORT.txt", ("\n".join(summaries) or "No Android Auto sessions have been logged yet.\n").encode())
config = identity_store.load_config(config_path)
add("config.json", json.dumps(shareable_config(config), indent=2, default=str).encode())
for name in EXTRA_LOGS:
data = _tail(directory / name, EXTRA_LOG_BYTES)
if data is not None:
add(f"logs/{name}", data)
return output.getvalue()
@@ -1,21 +1,97 @@
"""Read-only headless producer for a separate large StarPilot projection view.
Frames are rendered only while requested. Projection does not receive touch input.
Onroad, each new camera frame is drawn exactly once, as soon as it lands (see
CameraPacer). When the encoder takes NV12, the frame is converted on the GPU,
and it can be read back without stalling the renderer.
"""
from __future__ import annotations
import argparse
from contextlib import ExitStack
import gc
import os
import signal
import time
from openpilot.starpilot.system.android_auto.frame_source import FrameProducer, FrameRequest, frame_bytes
from openpilot.starpilot.system.android_auto.frame_source import (FLAG_ASYNC_READBACK, FLAG_NV12, FORMAT_NV12, FORMAT_RGBA,
FrameProducer, FrameRequest, frame_bytes)
from openpilot.starpilot.system.android_auto.projection_geometry import projection_geometry
STARTUP_WAIT_SECONDS = 15.0
CAMERA_WAIT_STEP = 0.01 # s; keeps demand/stop checks responsive while waiting for the camera
CAMERA_MAX_GAP = 0.1 # s; a shown camera silent this long stops pacing (the encoder rate applies)
class CameraPacer:
"""Onroad, draw each frame of the camera on screen exactly once, as soon as it lands.
The road cameras and the driving model run at 20 Hz. A 30 fps timer redraws
every third frame for nothing and shows camera frames in an uneven 2-1
cadence; a 20 fps timer drifts against the camera clock and periodically
repeats one frame and skips the next. camerad hands the frame to VisionIPC
before it publishes the matching CameraState, so waking on that message means
CameraView's non-blocking recv already has the new frame.
Only the shown camera's CameraState is subscribed, conflated and never
deserialized: the pacer only needs to know that one arrived. When that camera
goes quiet, it stops pacing and the encoder's frame rate applies again.
"""
def __init__(self, sock_factory=None, stream_types=None):
if sock_factory is None:
import openpilot.cereal.messaging as messaging
def sock_factory(name):
poller = messaging.Poller()
return poller, messaging.sub_sock(name, poller=poller, conflate=True)
if stream_types is None:
from openpilot.cereal.visionipc import VisionStreamType as stream_types
self.state_for_stream = {int(stream_types.VISION_STREAM_NARROW_ROAD): "narrowRoadCameraState",
int(stream_types.VISION_STREAM_WIDE_ROAD): "wideRoadCameraState",
int(stream_types.VISION_STREAM_CABIN): "cabinCameraState"}
self._sock_factory = sock_factory
self._socks: dict[str, tuple] = {} # opened on first use, so an unshown camera costs nothing
self._last_arrival: dict[str, float] = {}
def wait(self, stream_type, now: float) -> bool:
"""True when a new frame of ``stream_type`` is ready, or when that camera is quiet; waits at most one step."""
state = self.state_for_stream.get(int(stream_type))
if state is None:
return True
if state not in self._socks:
self._socks[state] = self._sock_factory(state)
poller, sock = self._socks[state]
quiet = now - self._last_arrival.get(state, float("-inf")) >= CAMERA_MAX_GAP
# A quiet camera must not hold rendering back: just check whether it has resumed.
arrived = bool(poller.poll(0 if quiet else int(CAMERA_WAIT_STEP * 1000))) and sock.receive(non_blocking=True) is not None
if arrived:
self._last_arrival[state] = now
return arrived or quiet
def create_readback(factory, size: int, asynchronous: bool):
try:
return factory(size, asynchronous=asynchronous)
except (RuntimeError, OSError, AttributeError) as error:
if not asynchronous:
raise
print(f"Async readback unavailable, using synchronous readback: {error}", flush=True)
return factory(size, asynchronous=False)
def finish_readback(factory, readback):
try:
return readback, readback.finish()
except RuntimeError as error:
if not readback.asynchronous:
raise
size = readback.size
readback.close()
print(f"Async readback failed, using synchronous readback: {error}", flush=True)
return factory(size, asynchronous=False), None
def visible_geometry(request: FrameRequest) -> tuple[int, int, float, int, int]:
@@ -56,7 +132,7 @@ def run(frames_path: str) -> int:
signal.signal(signal.SIGTERM, stop)
signal.signal(signal.SIGINT, stop)
from openpilot.starpilot.system.android_auto.headless_egl import FrameReadback, HeadlessContext
from openpilot.starpilot.system.android_auto.gpu_nv12 import compose_rgba
from openpilot.starpilot.system.android_auto import gpu_nv12
from openpilot.starpilot.system.android_auto.projection_onroad import ProjectionOnroad
import pyray as rl
from openpilot.system.ui.lib.application import gui_app
@@ -91,36 +167,113 @@ def run(frames_path: str) -> int:
if not output.id:
raise RuntimeError('Projection output target unavailable')
resources.callback(rl.unload_render_texture, output)
readback = FrameReadback(frame_bytes(request.width, request.height, 0), asynchronous=False)
resources.callback(readback.close)
converter = None
if request.flags & FLAG_NV12:
try:
# The composed frame already carries the margins and the top-down flip,
# so it converts as is. Reading back NV12 moves 1.5 bytes per pixel
# instead of 4, and the encoder skips its own RGBA conversion on the CPU.
converter = gpu_nv12.Nv12Converter(request.width, request.height)
resources.callback(converter.close)
except Exception as error:
print(f"NV12 conversion unavailable, publishing RGBA: {error}", flush=True)
converter = None
pixel_format = FORMAT_NV12 if converter is not None else FORMAT_RGBA
readback = create_readback(FrameReadback, frame_bytes(request.width, request.height, pixel_format),
asynchronous=bool(request.flags & FLAG_ASYNC_READBACK))
resources.callback(lambda: readback.close())
rgba_regions = [(output.id, request.width, request.height, 0)]
pipeline = f"{'nv12' if converter is not None else 'rgba'}, {'async' if readback.asynchronous else 'sync'} readback"
print(f"car view pipeline: {pipeline}", flush=True)
from openpilot.starpilot.system.android_auto import identity as identity_store
from openpilot.starpilot.system.android_auto.render_profile import RenderSampler, RenderSummary
config = identity_store.load_config()
sampler = None
if config["render_profile"]:
sampler = RenderSampler(identity_store.LOG_DIR / "render_profile.txt", max_bytes=config["render_profile_kb"] * 1024)
sampler.start()
resources.callback(sampler.close)
summary = RenderSummary(time.monotonic())
camera_pacer = CameraPacer()
in_flight_ns = 0
def publish_readback() -> None:
"""Hand the frame read back last to android_autod."""
nonlocal readback
readback, pixels = finish_readback(FrameReadback, readback)
if pixels is None:
return
try:
producer.publish(request, pixels, in_flight_ns, pixel_format, advance=False)
finally:
readback.release()
# Everything built so far lives for the whole session. Frozen, it is never rescanned
# by the collector, whose full passes over the UI's objects stall a frame every few seconds.
gc.collect()
gc.freeze()
while not stopped and os.getppid() == parent:
now = time.monotonic()
pending = producer.pending_request(now)
if pending is None:
if readback.pending:
readback.release()
if sampler is not None:
sampler.rendering = False
context.pause()
time.sleep(0.05)
summary.reset(time.monotonic())
continue
if pending != request:
return 3 # supervisor starts a new renderer for the new geometry
captured_ns = time.monotonic_ns()
# The requested frame rate (the encoder's budget) always applies. Rendering
# only when it is due also keeps the schedule from running ahead of real time.
delay = producer.capture_delay(request, captured_ns)
if delay > 0:
if readback.pending:
publish_readback() # never hold a finished frame back just to pace the next one
if sampler is not None:
sampler.rendering = False
time.sleep(min(delay, 0.05))
continue
if layout.camera_stream is not None:
# Within that budget, draw as soon as the camera on screen has a new frame.
if readback.pending:
publish_readback()
if not camera_pacer.wait(layout.camera_stream, now):
if sampler is not None:
sampler.rendering = False
continue
captured_ns = time.monotonic_ns()
now = captured_ns / 1e9
if sampler is not None:
sampler.rendering = True
frame_began = time.monotonic()
context.begin_frame(now)
ui_state.update()
if sampler is not None:
sampler.onroad = ui_state.started
rl.begin_texture_mode(content)
try:
rl.clear_background(rl.BLACK)
layout.render()
finally:
rl.end_texture_mode()
compose_rgba(content.texture, output, request.margin_w, request.margin_h, fit=True)
readback.start([(output.id, request.width, request.height, 0)])
try:
producer.publish(request, readback.finish(), captured_ns)
finally:
readback.release()
if readback.pending:
# The previous frame, read back while this one was drawn: waiting any
# later only adds latency.
publish_readback()
gpu_nv12.compose_rgba(content.texture, output, request.margin_w, request.margin_h, fit=True)
regions = converter.convert(output.texture) if converter is not None else rgba_regions
producer.advance(request, captured_ns)
readback.start(regions)
in_flight_ns = captured_ns
if not readback.asynchronous:
publish_readback()
report = summary.frame_done(frame_began, time.monotonic())
if report is not None and sampler is not None:
sampler.summary = report
gui_app._frame += 1
return 0
+10 -4
View File
@@ -14,7 +14,13 @@ import socket
import threading
from collections.abc import Callable
AG_FEATURES = 0 # no three-way calling, no codec negotiation, no voice recognition
# Advertise what an ordinary phone with no call in progress offers. With +BRSF: 0 the 2025 Civic (Alps Alpine
# 8A501) finished the SLC, then dropped the link 1 ms after AT+BIA and redialled every 10 s (btmon, 2026-10-01).
# Bits: three-way calling, EC/NR, reject call, enhanced call status, enhanced call control, extended error codes,
# codec negotiation. No voice recognition, in-band ring or eSCO S4: there is never call audio behind this gateway.
AG_FEATURES = 1 | 2 | 32 | 64 | 128 | 256 | 512
SDP_FEATURES = 0x01 | 0x02 | 0x20 # three-way calling, EC/NR, wide band speech (pairs with codec negotiation)
HFP_VERSION = 0x0107
INDICATORS = ",".join(('("call",(0,1))', '("callsetup",(0-3))', '("service",(0-1))', '("signal",(0-5))',
'("roam",(0,1))', '("battchg",(0-5))', '("callheld",(0-2))'))
INDICATOR_VALUES = "0,0,1,5,0,5,0"
@@ -39,11 +45,11 @@ def respond(command: str) -> list[str]:
return ["+BIND: (1,2)", "OK"]
if cmd == "AT+BIND?":
return ["+BIND: 1,0", "+BIND: 2,0", "OK"]
if cmd.startswith(("ATD", "ATA", "AT+BLDN", "AT+BVRA", "AT+CHLD=")):
return ["ERROR"] # no telephony behind this gateway
if cmd.startswith(("ATD", "ATA", "AT+BLDN", "AT+BVRA", "AT+CHLD=", "AT+BCC")):
return ["ERROR"] # no telephony or call audio behind this gateway
if cmd.startswith(("AT+CMER", "AT+CLIP", "AT+CCWA", "AT+NREC", "AT+VGS", "AT+VGM", "AT+CMEE", "AT+BIA", "AT+COPS=",
"AT+CLCC", "AT+CNUM", "AT+BTRH", "AT+BAC", "AT+BIND=", "AT+BIEV", "AT+XAPL", "AT+IPHONEACCEV",
"AT+CHUP", "AT+XEVENT", "AT+CSRSF", "AT+APLSIRI", "AT+BCS", "AT+BCC", "AT+NREC")):
"AT+CHUP", "AT+XEVENT", "AT+CSRSF", "AT+APLSIRI", "AT+BCS")):
return ["OK"]
return ["ERROR"]
@@ -10,6 +10,7 @@ from __future__ import annotations
import json
import os
import re
import ssl
import stat
import tempfile
@@ -82,10 +83,13 @@ def load_identity(directory: Path | None = None, now: datetime | None = None) ->
expires.isoformat() if expires is not None else "unknown", days_left)
CONFIG_VERSION = 2
CONFIG_VERSION = 3
# Before config versions, every save wrote these defaults, which pinned projection
# at 12 fps / 4000 kbps. Unversioned files holding exactly them get today's defaults.
LEGACY_DEFAULTS = {"fps": 12, "bitrate_kbps": 4000}
# Before version 3 the car view could only publish RGBA, and every save wrote these off.
# They were never a choice, so older files get today's defaults for them.
V3_RESET = ("gpu_nv12", "async_readback")
DEFAULT_CONFIG = {
"config_version": CONFIG_VERSION,
@@ -101,9 +105,9 @@ DEFAULT_CONFIG = {
"fps": 0, # 0 = automatic (30 with hardware, 15 with software); otherwise a cap, 5-30
"bitrate_kbps": 6000,
"rate_control": "cbr", # hardware encoder: "cbr" holds the bitrate (easier on the car's Wi-Fi); "vbr" as before
"gpu_nv12": False, # current car view publishes RGBA
"async_readback": False, # current car view uses synchronous readback
"render_profile": False, # current car view has no persistent profile writer
"gpu_nv12": True, # car view: convert to the encoder's NV12 on the GPU (a third of the RGBA readback)
"async_readback": True, # car view: read frames back without stalling the renderer on the GPU
"render_profile": True, # car view: always-on sampling profile in logs/render_profile.txt (and .1.txt)
"render_profile_kb": 256, # size cap per render_profile file
"wifi_interface": "wlan0",
"device_name": "StarPilot",
@@ -123,6 +127,8 @@ def load_config(path: Path | None = None) -> dict:
version = stored.get("config_version")
if not isinstance(version, int) or version < 2:
stored = {key: value for key, value in stored.items() if LEGACY_DEFAULTS.get(key, object()) != value}
if not isinstance(version, int) or version < 3:
stored = {key: value for key, value in stored.items() if key not in V3_RESET}
config.update({key: value for key, value in stored.items() if key in DEFAULT_CONFIG and isinstance(value, type(DEFAULT_CONFIG[key]))})
except (OSError, ValueError):
pass
@@ -165,5 +171,16 @@ def expiry_warning(identity: Identity) -> str:
return ""
def session_log_order(path: Path) -> tuple[int, str]:
"""Sort key for session logs, oldest first.
Files are numbered, because the clock can read a date from months ago until it syncs;
ordering by the timestamp in the name put the newest session first. Unnumbered files
are from before numbering, so they are the oldest.
"""
match = re.fullmatch(r"session-(\d{6})-.*\.jsonl", path.name)
return (int(match[1]), path.name) if match else (-1, path.name)
def timestamp() -> str:
return time.strftime("%Y%m%d-%H%M%S")
@@ -34,6 +34,12 @@ NM_IP4_CONFIG_IFACE = "org.freedesktop.NetworkManager.IP4Config"
DEVICE_TYPE_WIFI = 2
ACTIVE_STATE_ACTIVATED = 2
ACTIVE_STATE_DEACTIVATED = 4
# The car sends its Wi-Fi details ~4 s after ignition, but a Honda's access point took ~55 s from ignition to appear
# (2026-09-30 drives). After a reported failed join that car stopped auto-launching projection until the driver tapped
# Android Auto, so wait for a slow access point rather than give up. The car's reported BSSID gets most of the time;
# the SSID-only retry covers head units that report the wrong radio.
JOIN_TIMEOUT_S = 120.0
BSSID_SHARE = 0.75
CONNECTION_ID = "starpilot-android-auto"
SECURITY_WPA3 = (32,)
@@ -121,7 +127,7 @@ class NetworkLease:
self.interface = self._get(fallback, NM_DEVICE_IFACE, "Interface")
return fallback
def acquire(self, credentials: WifiCredentials, timeout: float = 40.0, cancelled: Callable[[], bool] = lambda: False) -> str:
def acquire(self, credentials: WifiCredentials, timeout: float = JOIN_TIMEOUT_S, cancelled: Callable[[], bool] = lambda: False) -> str:
"""Join the projection network and return the comma's IPv4 address on it.
The BSSID the car reports is tried first (fast, unambiguous); if that
@@ -134,7 +140,8 @@ class NetworkLease:
last_error: NetworkError | None = None
for index, attempt in enumerate(attempts):
try:
return self._acquire_once(attempt, timeout / len(attempts), cancelled)
share = 1.0 if len(attempts) == 1 else (BSSID_SHARE if index == 0 else 1.0 - BSSID_SHARE)
return self._acquire_once(attempt, timeout * share, cancelled)
except NetworkError as error:
last_error = error
if cancelled() or index == len(attempts) - 1:
@@ -64,6 +64,7 @@ class ProjectionOnroad:
self.customization = customization
self._base_customization = None
self._projection_customization = None
self.camera_stream = None # the camera stream the last frame drew, or None
self.native = dependencies or native_dependencies()
native = self.native
if native.ui_state.projection_read_only is not True:
@@ -109,6 +110,7 @@ class ProjectionOnroad:
self.onroad.close()
def _camera_layer(self, rect, state):
self.camera_stream = self.camera.stream_type
self.camera.render_camera_model_layer(rect, road_style=state.customization['roadColors']['large'])
def _driver_monitor_layer(self, rect, state):
@@ -135,6 +137,7 @@ class ProjectionOnroad:
def render(self):
ui = self.native.ui_state
self.camera_stream = None
if ui.started:
state = self.adapter.build(self.native.shell_mode.ONROAD, now_ns=time.monotonic_ns()).onroad
if self.customization is not None:
@@ -0,0 +1,175 @@
"""Always-on sampling profile of the car renderer, in a size-capped file.
A daemon thread looks at the render thread's Python stack every INTERVAL
seconds (sys._current_frames: no tracing, so the renderer runs at full speed)
and counts, per function, how often it was running (self) or on the stack
(inclusive), plus the hottest source lines. Samples taken while the renderer
is idle (pacing, or the car showing its own screen) are only counted as idle.
Every WINDOW seconds the busiest entries of that window are appended to the
report file together with the renderer's latest ``render_stats`` (see RenderSummary). When the
file would grow past ``max_bytes`` it becomes ``<name>.1.txt`` (replacing the
previous one) and a new file starts, so at most twice ``max_bytes`` is kept.
The file outlives renderer restarts, unlike car_ui.log.
The sampler can only look while the render thread lets go of the GIL, which it
does in every raylib/GL call, so pure-Python work is credited to the next such
call in the innermost-frame sections. The "on the stack" section is accurate
per function and is the one to read for which part of the layout costs time.
Sampling runs at 25 samples a second.
"""
from __future__ import annotations
import os
import sys
import threading
import time
from collections import Counter
from pathlib import Path
INTERVAL = 0.04 # 25 samples a second
WINDOW = 60.0
TOP = 25
TOP_LINES = 15
SUMMARY_INTERVAL = 5.0
class RenderSummary:
"""Frame rate and mean frame time of the renderer while it renders, for the profile header."""
def __init__(self, now: float, interval: float = SUMMARY_INTERVAL):
self.interval = interval
self.reset(now)
def reset(self, now: float) -> None:
self.started, self.frames, self.frame_seconds = now, 0, 0.0
def frame_done(self, began: float, done: float) -> dict | None:
"""Count one frame; every ``interval`` returns the window's render_stats and starts a new one."""
self.frames += 1
self.frame_seconds += done - began
if done - self.started < self.interval:
return None
report = {"event": "render_stats", "fps": round(self.frames / (done - self.started), 1),
"frame_ms": round(1000 * self.frame_seconds / self.frames, 1)}
self.reset(done)
return report
class RenderSampler:
def __init__(self, path: Path, *, max_bytes: int = 256 * 1024, interval: float = INTERVAL, window: float = WINDOW,
clock=time.monotonic):
self.path, self.max_bytes = path, max_bytes
self.interval, self.window, self.clock = interval, window, clock
self.thread_id = threading.get_ident() # created on the render thread, which it samples
self.rendering = False # set by the render loop around each frame
self.onroad = False
self.summary: dict | None = None # latest render_stats
self._names: dict = {}
self._lock = threading.Lock()
self._stop = threading.Event()
self._worker = threading.Thread(target=self._run, name="car_ui_sampler", daemon=True)
self._reset()
def _reset(self) -> None:
self.started = self.clock()
self.samples = self.idle = self.onroad_samples = 0
self.self_counts: Counter = Counter()
self.total_counts: Counter = Counter()
self.line_counts: Counter = Counter()
self.param_samples: Counter = Counter()
def start(self) -> None:
self._worker.start()
def close(self) -> None:
self._stop.set()
if self._worker.is_alive():
self._worker.join(timeout=1.0)
self.flush()
def _name(self, code) -> str:
name = self._names.get(code)
if name is None:
name = self._names[code] = f"{Path(code.co_filename).name}:{code.co_firstlineno} {code.co_name}"
return name
def sample(self, frame) -> None:
"""Count one sample of the render thread's stack (``frame`` is its innermost frame)."""
with self._lock:
self.samples += 1
if not self.rendering:
self.idle += 1
return
self.onroad_samples += self.onroad
code = frame.f_code
self.line_counts[f"{Path(code.co_filename).name}:{frame.f_lineno} {code.co_name}"] += 1
self.self_counts[self._name(code)] += 1
seen = set()
param_seen = False
while frame is not None:
code = frame.f_code
# Attribute sampled waits to keys without wrapping every Params call.
# Only key names are captured; values (including credentials) never are.
if not param_seen and code.co_name in ("get", "get_bool") and code.co_filename.endswith("/common/params.py"):
key = frame.f_locals.get("key")
if isinstance(key, str):
self.param_samples[f"{code.co_name}({key})"] += 1
param_seen = True
name = self._name(frame.f_code)
if name not in seen:
seen.add(name)
self.total_counts[name] += 1
frame = frame.f_back
def _run(self) -> None:
while not self._stop.wait(self.interval):
frame = sys._current_frames().get(self.thread_id)
if frame is not None:
self.sample(frame)
del frame
if self.clock() - self.started >= self.window:
self.flush()
def report(self) -> str:
with self._lock:
busy = self.samples - self.idle
if busy <= 0:
return ""
state = "onroad" if self.onroad_samples * 2 >= busy else "offroad"
lines = [f"==== {time.strftime('%Y-%m-%d %H:%M:%S')} {self.clock() - self.started:.0f} s, {state} " +
f"({100 * self.onroad_samples // busy}% onroad), rendering {100 * busy // self.samples}% of {self.samples} samples ===="]
if self.summary:
lines.append("render_stats " + " ".join(f"{key}={value}" for key, value in self.summary.items() if key != "event"))
for title, counts, top in (("on the stack (time inside each function)", self.total_counts, TOP),
("innermost function (pure Python shows up at the next raylib/GL call)", self.self_counts, TOP),
("innermost line", self.line_counts, TOP_LINES)):
lines.append(f"-- {title}")
# Frames on every sample (the render loop and interpreter startup) say nothing.
shown = [(name, count) for name, count in counts.most_common() if count < busy or counts is not self.total_counts]
lines += [f"{100 * count / busy:6.1f}% {name}" for name, count in shown[:top]]
if self.param_samples:
lines.append("-- Params keys sampled on the render thread (% of busy samples, not read counts)")
lines += [f"{100 * count / busy:6.1f}% {name}" for name, count in self.param_samples.most_common(TOP)]
return "\n".join(lines) + "\n\n"
def flush(self) -> None:
text = self.report()
with self._lock:
self._reset()
if text:
self._write(text)
def _write(self, text: str) -> None:
try:
self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
try:
if self.path.stat().st_size + len(text) > self.max_bytes:
os.replace(self.path, self.path.with_name(f"{self.path.stem}.1{self.path.suffix}"))
except FileNotFoundError:
pass
with open(self.path, "a") as handle:
handle.write(text)
except OSError:
pass
@@ -25,7 +25,7 @@ from collections.abc import Callable
from dataclasses import dataclass
from openpilot.starpilot.system.android_auto.touch import InputConfig, TouchEvent, TouchMapper, parse_input_config
from openpilot.starpilot.system.android_auto.wire import field, json_fields, one, parse_fields, signed
from openpilot.starpilot.system.android_auto.wire import describe, describe_fields, field, json_fields, one, parse_fields, signed
MAX_MESSAGE = 2 * 1024 * 1024
PHONE_MAX_VERSION = (6, 1) # protocol version Android Auto 17.6 reports to newer head units
@@ -76,6 +76,14 @@ FOCUS_REASON_USER_SELECTION = 4
SHUTDOWN_REASON_USER_SELECTION = 1
RESOLUTIONS = {1: (800, 480), 2: (1280, 720), 3: (1920, 1080)}
# Named in errors and reports only: 1440p, 4K and portrait screens also offer 800x480, which is mandatory.
RESOLUTION_NAMES = {**{index: f"{w}x{h}" for index, (w, h) in RESOLUTIONS.items()}, 4: "2560x1440", 5: "3840x2160",
6: "720x1280", 7: "1080x1920", 8: "1440x2560", 9: "2160x3840"}
CODEC_NAMES = {3: "H.264", 5: "VP9", 6: "AV1", 7: "H.265"}
# Messages we do not handle are logged with a readable body, but only the first few of each
# kind and then every hundredth, so a head unit repeating one cannot flood the session log.
IGNORED_LOG_FIRST = 5
IGNORED_LOG_EVERY = 100
FRAME_RATES = {1: 60, 2: 30}
# Software H.264 on the comma is the constraint: prefer 720p, then 480p. 1080p
# is accepted only when nothing smaller is offered.
@@ -131,6 +139,25 @@ class VideoMode:
"fps": self.fps, "margin_width": self.margin_width, "margin_height": self.margin_height}
def describe_video_config(config: dict, display_type: int | None = None) -> str:
"""One offered video configuration (parsed fields), e.g. "1280x720 H.264 margins 0x240 (display 1)"."""
resolution, codec = one(config, 1), one(config, 10, CODEC_H264_BP)
text = f"{RESOLUTION_NAMES.get(resolution, f'resolution {resolution}')} {CODEC_NAMES.get(codec, f'codec {codec}')}"
margin_width, margin_height = one(config, 3, 0) or 0, one(config, 4, 0) or 0
if margin_width or margin_height:
text += f" margins {margin_width}x{margin_height}"
if display_type:
text += f" (display {display_type})"
return text
def describe_video_configs(channels: list[dict]) -> str:
"""What the head unit offered, for an error a person can act on."""
offered = [describe_video_config(config, channel.get("display_type"))
for channel in channels for config in channel.get("video_configs", [])]
return ", ".join(offered) or "no video"
def choose_video_mode(channels: list[dict]) -> VideoMode:
"""Pick a negotiated H.264 mode the software encoder can sustain."""
candidates = []
@@ -148,7 +175,7 @@ def choose_video_mode(channels: list[dict]) -> VideoMode:
mode = VideoMode(int(channel["id"]), index, width, height, fps, margin_width, margin_height)
candidates.append((RESOLUTION_PREFERENCE.index(resolution), fps != 30, index, mode))
if not candidates:
raise ValueError("Head unit did not advertise a supported H.264 video mode")
raise ValueError(f"Head unit did not advertise a supported H.264 video mode (offered {describe_video_configs(channels)})")
return min(candidates, key=lambda item: item[:3])[3]
@@ -178,6 +205,7 @@ class Session:
self.fragments: dict[int, tuple[int, int, bytearray]] = {}
self.bytes_sent = 0
self.bytes_received = 0
self.ignored_counts: dict[tuple[str, int, int], int] = {}
@staticmethod
def _timeout(value: float) -> float:
@@ -190,6 +218,13 @@ class Session:
if self._log is not None:
self._log(name, **values)
def ignored(self, name: str, channel: int, kind: int, data: bytes, **values) -> None:
"""Log a message this side does not act on, rate-limited per (event, channel, kind)."""
key = (name, channel, kind)
count = self.ignored_counts[key] = self.ignored_counts.get(key, 0) + 1
if count <= IGNORED_LOG_FIRST or count % IGNORED_LOG_EVERY == 0:
self.event(name, channel=channel, kind=kind, count=count, bytes=len(data), message=describe(data), **values)
# ----------------------------------------------------------------- framing
def send(self, channel: int, kind: int, body: bytes = b"", encrypted: bool = True, control: bool = False, *,
@@ -354,6 +389,10 @@ class Session:
done = True
except ssl.SSLWantReadError:
pass
except ssl.SSLError as error:
# e.g. NO_SHARED_CIPHER from an old head-unit TLS stack, or a certificate the root does not verify
self.event("tls_failed", reason=error.reason or "", library=error.library or "", error=str(error)[:200])
raise
response = self.outgoing.read()
if response:
self.send(0, MSG_SSL_HANDSHAKE, response, encrypted=False)
@@ -390,24 +429,30 @@ class Session:
if not isinstance(descriptor, bytes):
continue
fields = parse_fields(descriptor)
item: dict = {"id": one(fields, 1), "services": sorted(number for number in fields if number != 1)}
item: dict = {"id": one(fields, 1), "services": sorted(number for number in fields if number != 1),
"descriptor": describe_fields(fields)}
av = one(fields, 3)
if isinstance(av, bytes):
media = parse_fields(av)
item["media_type"] = one(media, 1)
item["display_id"], item["display_type"] = one(media, 6), one(media, 7)
item["video_configs"] = [parse_fields(c) for c in media.get(4, []) if isinstance(c, bytes)]
if isinstance(one(fields, 4), bytes):
item["input"] = True
try:
item["display_id"] = one(parse_fields(one(fields, 4)), 5) # the display this input belongs to
except ValueError:
pass
try:
item["input_config"] = parse_input_config(one(fields, 4))
except ValueError:
item["input_config"] = InputConfig()
channels.append(item)
head_unit = {number: [value.decode("utf-8", "replace") for value in values if isinstance(value, bytes)]
for number, values in response.items() if number in (2, 3, 4, 5, 6, 7, 8, 9)}
# Everything but the channel list: make, model, year, software and, on newer units, headunit_info.
head_unit = redact_head_unit(describe_fields({number: values for number, values in response.items() if number != 1}))
self.event("discovered", channels=[{k: (json_fields_list(v) if k == "video_configs" else str(v) if k == "input_config" else v)
for k, v in ch.items()} for ch in channels],
head_unit={k: v for k, v in head_unit.items() if v})
head_unit=head_unit)
return channels
def wait_for(self, channel: int, kind: int, timeout: float = 10.0) -> bytes:
@@ -424,7 +469,22 @@ class Session:
if channel == 0 and kind == MSG_PING_REQUEST:
self.send(0, MSG_PING_RESPONSE, field(1, one(parse_fields(data), 1, 0)))
elif not (channel == 0 and kind == MSG_PING_RESPONSE):
self.event("unexpected_while_waiting", channel=channel, kind=kind, expected=expected, bytes=len(data))
self.ignored("unexpected_while_waiting", channel, kind, data, expected=expected)
SDR_VEHICLE_ID = 5 # ServiceDiscoveryResponse.vehicle_id
HEAD_UNIT_INFO = 17 # ServiceDiscoveryResponse.headunit_info
HEAD_UNIT_INFO_VEHICLE_ID = 4
def redact_head_unit(described: dict) -> dict:
"""Drop the vehicle identifier from a described discovery response: logs are shared for diagnosis."""
if SDR_VEHICLE_ID in described:
described[SDR_VEHICLE_ID] = ["redacted"]
for info in described.get(HEAD_UNIT_INFO, []):
if isinstance(info, dict) and HEAD_UNIT_INFO_VEHICLE_ID in info:
info[HEAD_UNIT_INFO_VEHICLE_ID] = ["redacted"]
return described
def json_fields_list(configs: list[dict]) -> list[dict]:
@@ -434,7 +494,11 @@ def json_fields_list(configs: list[dict]) -> list[dict]:
class ProjectionSession(Session):
"""Video projection with focus epochs, bounded ACK window and mandatory resume keyframes."""
ACK_TIMEOUT = 1.5 # Wi-Fi adds jitter compared to the donor's USB 500 ms budget.
# Wi-Fi jitter, plus bursts where this loop itself stalls: the dashcam encoders and the model starting at onroad
# held it ~1 s on 2026-10-01 while the car kept acking in 27-219 ms. Stays under the 3 s after which a Honda takes
# its screen back without video.
ACK_TIMEOUT = 2.5
ACK_DRAIN_SECONDS = 0.1 # before calling an ACK late, read what already arrived, for at most this long
MAX_WINDOW = 2
def __init__(self, *args, **kwargs):
@@ -455,6 +519,8 @@ class ProjectionSession(Session):
self.focus_epoch = 0
self.max_ack_seconds = 0.0
self.config_ack_slack = 0 # codec-config messages a head unit may acknowledge like frames
self.epoch_acked = 0
self.video_confirmed = False
self.input_channel: int | None = None
self.input_events = 0
self.touch: TouchMapper | None = None
@@ -483,7 +549,7 @@ class ProjectionSession(Session):
self.send(mode.channel, AV_SETUP_REQUEST, field(1, CODEC_H264_BP))
setup = parse_fields(self.wait_for(mode.channel, AV_SETUP_RESPONSE))
window = int(one(setup, 2, 0) or 0)
if one(setup, 1) != SETUP_STATUS_READY or not 1 <= window <= 32:
if one(setup, 1) != SETUP_STATUS_READY or window < 1: # the window is capped to MAX_WINDOW below; MIB3 offers 100
raise ValueError(f"Unsupported video setup response: {json_fields(setup)}")
configs = setup.get(3, [])
if configs and mode.config_index not in configs:
@@ -556,14 +622,14 @@ class ProjectionSession(Session):
self.event("peer_requested_shutdown", reason=one(fields, 1))
raise PeerRequestedStop(f"Head unit ended projection (reason {one(fields, 1)})")
else:
self.event("control_ignored", kind=kind, fields=json_fields(fields))
self.ignored("control_ignored", channel, kind, data)
elif mode is not None and channel == mode.channel:
if kind == VIDEO_FOCUS_INDICATION:
self._handle_focus(one(fields, 1), one(fields, 2, 0))
elif kind == AV_MEDIA_ACK:
self._handle_ack(one(fields, 1), int(one(fields, 2, 0) or 0))
else:
self.event("video_ignored", kind=kind, fields=json_fields(fields))
self.ignored("video_ignored", channel, kind, data)
elif channel == self.input_channel:
if kind == INPUT_EVENT:
self.input_events += 1
@@ -575,9 +641,9 @@ class ProjectionSession(Session):
elif kind == INPUT_BINDING_RESPONSE:
self.event("input_bound", status=signed(one(fields, 1, 0)))
else:
self.event("input_ignored", kind=kind)
self.ignored("input_ignored", channel, kind, data)
else:
self.event("channel_ignored", channel=channel, kind=kind, bytes=len(data))
self.ignored("channel_ignored", channel, kind, data)
def _handle_focus(self, focus, unsolicited) -> None:
was_focused = self.focused
@@ -596,6 +662,7 @@ class ProjectionSession(Session):
self.unacked = 0
self.pending.clear()
self.config_ack_slack = 0
self.epoch_acked = 0
self.media_started = True
self.needs_keyframe = True
self.focus_epoch += 1
@@ -608,12 +675,14 @@ class ProjectionSession(Session):
# session (as an extra count) or, like the DHU, under session 0.
if sid != self.session_id and 0 < count <= self.config_ack_slack:
self.config_ack_slack -= count
self._confirm_video()
return
excess = count - self.unacked
if sid == self.session_id and 0 < excess <= self.config_ack_slack:
self.config_ack_slack -= excess
count -= excess
if count == 0:
self._confirm_video()
return
if sid != self.session_id or not 0 < count <= self.unacked or count > len(self.pending):
raise ValueError(f"Invalid video acknowledgement session={sid} count={count} pending={self.unacked}")
@@ -622,14 +691,36 @@ class ProjectionSession(Session):
self.max_ack_seconds = max(self.max_ack_seconds, now - self.pending.popleft())
self.unacked -= count
self.acked += count
self.epoch_acked += count
self._confirm_video()
def _confirm_video(self) -> None:
# Same-session ACKs do not distinguish configuration from frames. Until the
# optional configuration ACKs are accounted for, use the conservative lower
# bound on acknowledged frames without changing flow-control credit.
if not self.video_confirmed and self.epoch_acked > self.config_ack_slack:
self.video_confirmed = True
self.event("video_acknowledged", session=self.session_id, ack_ms=round(self.max_ack_seconds * 1000))
def can_send(self) -> bool:
return self.focused and self.unacked < self.window
def _ack_overdue(self, now: float) -> bool:
return bool(self.focused and self.pending and now - self.pending[0] > self.ACK_TIMEOUT)
def check_progress(self, now: float | None = None) -> None:
now = time.monotonic() if now is None else now
if self.focused and self.pending and now - self.pending[0] > self.ACK_TIMEOUT:
raise TimeoutError(f"Video acknowledgement older than {self.ACK_TIMEOUT:.1f} s")
measured = now is None
now = time.monotonic() if measured else now
if self._ack_overdue(now):
# ACKs that arrived while this loop was busy (a slow encode, a CPU burst) still sit in the socket; the
# streaming loop drains only a bounded batch per pass, so read what is there before calling the car silent.
deadline = time.monotonic() + self.ACK_DRAIN_SECONDS
while time.monotonic() < deadline and self.pump(0.0):
pass
if measured:
now = time.monotonic()
if self._ack_overdue(now):
raise TimeoutError(f"Video acknowledgement older than {self.ACK_TIMEOUT:.1f} s")
# Only a receiver known to answer pings can be declared dead by silence; for
# others a lost link surfaces as a socket error or a dropped Wi-Fi lease.
if self.ping_responses and now - self.last_rx > 15.0:
@@ -18,7 +18,7 @@ from __future__ import annotations
import json
import os
import re
import queue
import socket
import threading
import time
@@ -47,9 +47,11 @@ UNAVAILABLE_AFTER = 1.0 # focused but no fresh UI frame for this long -> "u
SDP_SETTLE = (1.5, 2.2, 3.0)
TCP_ATTEMPTS = 6
MAX_LOG_FILES = 20
LOG_QUEUE_MAX = 1000 # session log records waiting for the writer; past this they are dropped and counted
ENCODER_RECOVERIES = 3 # hardware encoder reopens allowed per window before the session is torn down
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"
WIRED_USB_VERIFIED = False # configfs gadget/vehicle CAN USB isolation has not been proven on target
@@ -101,51 +103,84 @@ class UsbLease:
class EventLog:
"""Sanitized JSONL session log under /data/android_auto/logs, plus the recent tail in memory."""
"""Sanitized JSONL session log under /data/android_auto/logs, plus the recent tail in memory.
The streaming loop logs too, so records are queued and a background thread writes them: a slow
or full disk never stalls projection. If the writer falls behind, records are dropped and the
next one that fits is preceded by a ``log_dropped`` count.
"""
def __init__(self, directory: Path | None = None):
self.directory = directory or identity_store.LOG_DIR
self.handle = None
self.recent: deque[dict] = deque(maxlen=40)
self.lock = threading.Lock()
@staticmethod
def _order(path: Path) -> tuple[int, str]:
# Files are numbered, because the clock can read a date from months ago until it syncs;
# pruning by the timestamp in the name deleted the newest session first.
# Unnumbered files are from before numbering, so they are the oldest.
match = re.fullmatch(r"session-(\d{6})-.*\.jsonl", path.name)
return (int(match[1]), path.name) if match else (-1, path.name)
self._queue: queue.Queue = queue.Queue(maxsize=LOG_QUEUE_MAX)
self._writer: threading.Thread | None = None
self._writer_stop: threading.Event | None = None
self._dropped = 0
def open(self) -> None:
try:
self.directory.mkdir(mode=0o700, parents=True, exist_ok=True)
logs = sorted(self.directory.glob("session-*.jsonl"), key=self._order)
logs = sorted(self.directory.glob("session-*.jsonl"), key=identity_store.session_log_order)
for old in logs[:max(0, len(logs) - MAX_LOG_FILES + 1)]:
old.unlink(missing_ok=True)
number = max([0, *(self._order(log)[0] for log in logs)]) + 1
number = max([0, *(identity_store.session_log_order(log)[0] for log in logs)]) + 1
path = self.directory / f"session-{number:06d}-{identity_store.timestamp()}.jsonl"
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
self.handle = os.fdopen(fd, "a")
handle = os.fdopen(fd, "a")
except OSError:
self.handle = None
return
with self.lock:
self._queue = queue.Queue(maxsize=LOG_QUEUE_MAX)
self._dropped = 0
self._writer_stop = threading.Event()
self._writer = threading.Thread(target=self._write, args=(handle, self._queue, self._writer_stop),
name="aa_session_log", daemon=True)
self._writer.start()
@staticmethod
def _write(handle, records: queue.Queue, stopped: threading.Event) -> None:
with handle:
while not stopped.is_set() or not records.empty():
try:
line = records.get(timeout=0.1)
except queue.Empty:
continue
try:
handle.write(line)
if records.empty():
handle.flush()
except OSError:
pass
def close(self) -> None:
"""Write what is queued, then stop the writer (bounded: a stuck disk is left to the daemon thread)."""
with self.lock:
if self.handle is not None:
self.handle.close()
self.handle = None
writer, stopped = self._writer, self._writer_stop
self._writer = self._writer_stop = None
if writer is None:
return
assert stopped is not None
stopped.set()
writer.join(timeout=2.0)
def _line(self, record: dict) -> str:
return json.dumps(record, default=str) + "\n"
def __call__(self, name: str, **values) -> None:
record = {"t": datetime.now(UTC).isoformat(timespec="milliseconds"), "event": name, **values}
with self.lock:
self.recent.append(record)
if self.handle is not None:
try:
self.handle.write(json.dumps(record, default=str) + "\n")
self.handle.flush()
except OSError:
pass
if self._writer is None:
return
try:
if self._dropped:
self._queue.put_nowait(self._line({"t": record["t"], "event": "log_dropped", "count": self._dropped}))
self._dropped = 0
self._queue.put_nowait(self._line(record))
except queue.Full:
self._dropped += 1
class Supervisor:
@@ -173,6 +208,7 @@ class Supervisor:
self._pairing_until = 0.0
self._pairing_known: set[str] | None = None # Android Auto cars already paired when the pairing window opened
self._car_seen_at = -CAR_LINK_HOLD
self._hfp_link = threading.Event() # set when the chosen car opens the hands-free link
self.auto = AutoConnectPolicy()
self._companion_retry_at = 0.0
self.log = EventLog()
@@ -589,6 +625,7 @@ class Supervisor:
def _hfp_connected(self, address: str) -> None:
if address.upper() in (self.config["receiver_address"].upper(), self.config.get("companion_address", "").upper()):
self._car_seen_at = time.monotonic()
self._hfp_link.set()
def _release_phone(self) -> None:
"""Stop looking like a phone; keep only the standby gateway auto-connect needs."""
@@ -667,6 +704,30 @@ class Supervisor:
if self._stop.wait(seconds):
raise Cancelled()
def _car_link_fresh(self) -> bool:
return time.monotonic() - self._car_seen_at < HFP_FRESH
def _wait_backoff(self, delay: float, car_can_wake: bool) -> None:
"""Sleep out a retry delay, but retry at once when the car opens hands-free to us.
That is the car reaching out (at startup, or when the driver taps Android Auto); it hangs
up again within a second, so waiting out a 30 s backoff would miss it.
"""
if not car_can_wake:
self._wait(delay)
return
if not self._car_link_fresh():
self._hfp_link.clear()
deadline = time.monotonic() + delay
while (remaining := deadline - time.monotonic()) > 0:
if self._hfp_link.is_set():
# One early retry per connection from the car: if that attempt also fails fast,
# the next backoff must not end at once just because the link is still fresh.
self._hfp_link.clear()
self.log("retry_early", reason="car opened hands-free", skipped_s=round(remaining, 1))
return
self._wait(min(0.1, remaining))
# ---------------------------------------------------------------- session
def _run(self, generation: int, trigger: str = "manual") -> None:
@@ -709,7 +770,7 @@ class Supervisor:
delay = max(delay, PEER_STOP_RETRY_SECONDS) # the car ended projection itself; do not bounce straight back
self._set(state="backoff", attempt=attempt, retry_in=delay, mode=None)
try:
self._wait(delay)
self._wait_backoff(delay, car_can_wake=not wired and not peer_stopped)
except Cancelled:
break
finally:
@@ -92,6 +92,7 @@ class FakeHeadUnit:
self.start_indications: list[int] = []
self.device_name = ""
self.shutdown_received = threading.Event()
self.hold_acks = threading.Event() # set: receive video frames but stop acknowledging them
self.streaming = threading.Event()
self.stop = threading.Event()
self.error: BaseException | None = None
@@ -226,8 +227,9 @@ class FakeHeadUnit:
self.frames.append((session_id, data[8:]))
unacked += 1
self.streaming.set()
self._send(3, 0x8004, field(1, session_id) + field(2, 1))
unacked -= 1
if not self.hold_acks.is_set():
self._send(3, 0x8004, field(1, session_id) + field(2, 1))
unacked -= 1
elif channel == 1 and kind == 0x8002:
self._send(1, 0x8003, field(1, 0))
@@ -250,8 +252,15 @@ class FakeHeadUnit:
def rfcomm_head_unit(sock: socket.socket, endpoint: tuple[str, int], *, version_first: bool = False, oaa_layout: bool = False,
setup_info_only: bool = False, pings: bool = True, byte_by_byte: bool = False,
wait_for_phone_start: bool = False) -> dict:
"""Car side of the RFCOMM bootstrap; returns what the phone sent."""
wait_for_phone_start: bool = False, answer_start: bool = False, busy_replies: int = 0,
busy_follow_up: bool = False) -> dict:
"""Car side of the RFCOMM bootstrap; returns what the phone sent.
``answer_start`` models a 2025 Honda: it answers the phone's WifiStartRequest with
WifiStartResponse(ip, port, status) instead of sending its own request, after
``busy_replies`` "not ready" answers (status alone). With ``busy_follow_up`` it sends
the endpoint unprompted after the busy answer; otherwise it waits to be asked again.
"""
seen: dict = {"messages": []}
reader = bs.FrameReader()
queue: list[tuple[int, bytes]] = []
@@ -288,13 +297,23 @@ def rfcomm_head_unit(sock: socket.socket, endpoint: tuple[str, int], *, version_
message_id, payload = receive()
assert message_id == bs.WIFI_VERSION_RESPONSE
seen["version_response"] = parse_fields(payload)
if wait_for_phone_start: # 2025 Honda: projection starts only when the phone asks
message_id, payload = receive()
assert message_id == bs.WIFI_START_REQUEST, message_id
seen["phone_start_request"] = payload
if wait_for_phone_start: # 2025 Honda: projection starts only when the phone asks; some peers skip the version
message_id, payload = receive()
assert message_id == bs.WIFI_START_REQUEST, message_id
seen["phone_start_request"] = payload
if pings:
send(bs.WIFI_PING_REQUEST, field(1, 123))
send(bs.WIFI_START_REQUEST, field(1, ip) + field(2, port))
if answer_start:
for _ in range(busy_replies):
send(bs.WIFI_START_RESPONSE, field(3, -1))
if not busy_follow_up:
message_id, _ = receive()
while message_id == bs.WIFI_PING_RESPONSE:
message_id, _ = receive()
assert message_id == bs.WIFI_START_REQUEST, message_id
send(bs.WIFI_START_RESPONSE, field(1, ip) + field(2, port) + field(3, 0))
else:
send(bs.WIFI_START_REQUEST, field(1, ip) + field(2, port))
message_id, payload = receive()
while message_id == bs.WIFI_PING_RESPONSE:
message_id, payload = receive()
@@ -103,7 +103,11 @@ def test_sdp_channel_encodings():
# ----------------------------------------------------------------- bootstrap
def run_bootstrap(start_request_delay=5.0, **hu_options):
def unexpected_join(_):
raise AssertionError("Unexpected join")
def run_bootstrap(start_request_delay=5.0, initial_kick_delay=bs.INITIAL_KICK_SECONDS, **hu_options):
phone, car = socket.socketpair()
joined = []
result_holder = {}
@@ -118,7 +122,7 @@ def run_bootstrap(start_request_delay=5.0, **hu_options):
thread.start()
events = []
boot = bs.WirelessBootstrap(phone, lambda name, **values: events.append((name, values)), stage_timeout=5.0,
start_request_delay=start_request_delay)
start_request_delay=start_request_delay, initial_kick_delay=initial_kick_delay)
def join(credentials):
time.sleep(0.3) # the car pings while we join
@@ -155,6 +159,117 @@ def test_bootstrap_asks_car_to_start_projection():
assert ("bootstrap_tx", {"message": "WifiStartRequest", "bytes": 0}) in events
def test_bootstrap_takes_endpoint_from_start_response():
# 2025 Honda (session 25): version exchange, phone asks, car answers with a StartResponse.
result, joined, seen, events = run_bootstrap(start_request_delay=0.1, version_first=True, wait_for_phone_start=True,
answer_start=True)
assert (result.endpoint.ip, result.endpoint.port) == ("192.168.50.1", 5288) and joined[0].ssid == "HondaAA"
assert seen["messages"].count(bs.WIFI_START_REQUEST) == 1
assert not any(name == "bootstrap_ignored" for name, _ in events)
def test_bootstrap_waits_out_a_busy_start_response():
# The car says "not ready" (status alone), then sends the endpoint on its own.
result, joined, _, events = run_bootstrap(start_request_delay=0.1, version_first=True, wait_for_phone_start=True,
answer_start=True, busy_replies=1, busy_follow_up=True)
assert result.endpoint.port == 5288 and joined[0].ssid == "HondaAA"
assert ("bootstrap_start_refused", {"status": -1, "endpoint": False}) in events
def test_bootstrap_asks_again_after_a_busy_start_response():
result, joined, seen, _ = run_bootstrap(start_request_delay=0.1, version_first=True, wait_for_phone_start=True,
answer_start=True, busy_replies=2)
assert result.endpoint.port == 5288 and joined[0].ssid == "HondaAA"
assert seen["messages"].count(bs.WIFI_START_REQUEST) == 3
def test_bootstrap_start_response_without_endpoint_is_not_an_endpoint(monkeypatch):
now = [0.0]
monkeypatch.setattr(bs.time, "monotonic", lambda: now[0])
boot = bs.WirelessBootstrap(None, lambda *a, **k: None, stage_timeout=1, start_request_delay=0.3)
sent = []
monkeypatch.setattr(boot, "send", lambda message, payload=b"": sent.append(message))
replies = [(bs.WIFI_START_RESPONSE, field(3, 0)), (bs.WIFI_START_RESPONSE, field(1, "10.0.0.1") + field(2, 5288) + field(3, -3))]
def receive(timeout):
now[0] += 0.2
if replies:
return replies.pop(0)
raise bs.BootstrapTimeout("wifi_start", "head unit did not answer in time")
monkeypatch.setattr(boot, "next_frame", receive)
with pytest.raises(bs.BootstrapTimeout):
boot.run(unexpected_join)
assert bs.WIFI_START_REQUEST in sent, "a refused start is asked again"
def test_bootstrap_silent_peer_receives_one_prompt():
_, joined, seen, events = run_bootstrap(initial_kick_delay=0.05, wait_for_phone_start=True, pings=False)
assert joined[0].ssid == "HondaAA"
assert seen["messages"].count(bs.WIFI_START_REQUEST) == 1
assert sum(name == "bootstrap_initial_kick" for name, _ in events) == 1
def test_bootstrap_version_prompt_does_not_get_an_extra_initial_prompt():
_, _, seen, events = run_bootstrap(initial_kick_delay=0.05, start_request_delay=0.1,
version_first=True, wait_for_phone_start=True)
assert seen["messages"].count(bs.WIFI_START_REQUEST) == 1
assert not any(name == "bootstrap_initial_kick" for name, _ in events)
@pytest.mark.parametrize("stage", ["wifi_start", "wifi_info"])
def test_bootstrap_stage_deadline_survives_continuous_pings(monkeypatch, stage):
now = [0.0]
monkeypatch.setattr(bs.time, "monotonic", lambda: now[0])
boot = bs.WirelessBootstrap(None, lambda *a, **k: None, stage_timeout=1)
sent = []
monkeypatch.setattr(boot, "send", lambda message, payload=b"": sent.append(message))
first = [True]
def receive(timeout):
now[0] += 0.2
if first[0] and stage == "wifi_info":
first[0] = False
return bs.WIFI_START_REQUEST, field(1, "10.0.0.1") + field(2, 5288)
return bs.WIFI_PING_REQUEST, b""
monkeypatch.setattr(boot, "next_frame", receive)
with pytest.raises(bs.BootstrapTimeout) as error:
boot.run(unexpected_join)
assert error.value.stage == stage and now[0] < 1.5
assert bs.WIFI_PING_RESPONSE in sent and bs.WIFI_START_REQUEST not in sent
def test_bootstrap_hint_cannot_hide_disconnect():
phone, car = socket.socketpair()
try:
car.sendall(bs.encode_frame(bs.WIFI_SETUP_INFO, field(4, field(1, "10.0.0.1") + field(2, 5288))))
car.close()
with pytest.raises(bs.BootstrapError, match="closed"):
bs.WirelessBootstrap(phone, lambda *a, **k: None).run(unexpected_join)
finally:
phone.close()
car.close()
def test_bootstrap_hint_grace_is_not_extended_by_pings(monkeypatch):
now = [0.0]
monkeypatch.setattr(bs.time, "monotonic", lambda: now[0])
boot = bs.WirelessBootstrap(None, lambda *a, **k: None, stage_timeout=10)
monkeypatch.setattr(boot, "send", lambda *a: None)
payload = field(4, field(1, "10.0.0.1") + field(2, 5288)) + field(5, field(1, "test") + field(3, "secret-key") + field(4, 8))
def receive(timeout):
now[0] += 0.25
return (bs.WIFI_SETUP_INFO, payload) if now[0] == 0.25 else (bs.WIFI_PING_REQUEST, b"")
monkeypatch.setattr(boot, "next_frame", receive)
joined = []
result = boot.run(joined.append)
assert result.endpoint.ip == "10.0.0.1" and joined[0].ssid == "test"
assert 3 <= now[0] < 5
def test_bootstrap_detects_alternate_info_layout():
_, joined, _, _ = run_bootstrap(oaa_layout=True)
assert joined[0].key == "secret-key" and joined[0].bssid == "AA:BB:CC:DD:EE:FF"
@@ -211,7 +326,10 @@ def test_frame_reader_bounds():
# ----------------------------------------------------------------------- hfp
def test_hfp_service_level_connection():
assert hfp.respond("AT+BRSF=1015") == ["+BRSF: 0", "OK"]
assert hfp.respond("AT+BRSF=767") == ["+BRSF: 995", "OK"] # a phone's feature set, not "nothing"
assert hfp.respond("AT+BAC=1,2") == ["OK"]
assert hfp.respond("AT+BCC") == ["ERROR"] # never start call audio we cannot carry
assert hfp.respond("AT+BIA=1,1,1,1,1,1,1,0") == ["OK"]
assert hfp.respond("AT+CIND=?")[0].startswith("+CIND: (\"call\"")
assert hfp.respond("AT+CIND?") == ["+CIND: 0,0,1,5,0,5,0", "OK"]
assert hfp.respond("AT+CMER=3,0,0,1") == ["OK"]
@@ -227,7 +345,7 @@ def test_hfp_serve_over_socket():
b.sendall(b"AT+BRSF=0\rAT+CIND=?\r")
time.sleep(0.2)
data = b.recv(4096)
assert b"+BRSF: 0" in data and data.count(b"OK") == 2
assert b"+BRSF: 995" in data and data.count(b"OK") == 2
b.close()
thread.join(2)
stop.set()
@@ -445,6 +563,73 @@ def test_session_keeps_unsolicited_focus_grant(identity):
assert hu.error is None and hu.start_indications == [1]
HONDA_SCREEN_TAKEBACK_S = 3.0 # the Civic gives its screen back after this long without video
def streaming_session(hu: FakeHeadUnit, identity) -> ProjectionSession:
session = connect(hu, identity)
session.authenticate()
session.start("StarPilot", "comma.ai")
pump_until(session, lambda: session.focused)
session.send_frame(keyframe_au(1), 1, keyframe=True)
pump_until(session, lambda: session.acked == 1 and not session.pending)
# ACKs are counts, not frame ids: read everything the car sent (the codec config's ACK too) so a late one is not
# credited to the next frame.
deadline = time.monotonic() + 5
while session.pump(0.2):
assert time.monotonic() < deadline, "the car kept talking"
return session
def sent_frame(hu: FakeHeadUnit, session: ProjectionSession) -> float:
"""Send one frame and wait until the car has it; its ACK, if any, is left unread in the socket."""
received = len(hu.frames)
session.send_frame(delta_au(2), 2, keyframe=False)
deadline = time.monotonic() + 5
while len(hu.frames) == received:
assert time.monotonic() < deadline, "the car never received the frame"
time.sleep(0.01)
time.sleep(0.1)
return session.pending[0]
def end_session(session: ProjectionSession, hu: FakeHeadUnit) -> None:
hu.close()
session.peer.close()
def test_ack_timeout_leaves_room_for_a_stall_but_beats_the_car_taking_its_screen_back():
assert 1.5 < ProjectionSession.ACK_TIMEOUT < HONDA_SCREEN_TAKEBACK_S
def test_ack_already_received_during_a_loop_stall_keeps_the_session(identity):
# 2026-10-01: the dashcam and model starting at onroad stalled the comma's loop while the car kept acking.
hu = FakeHeadUnit(identity)
session = streaming_session(hu, identity)
sent = sent_frame(hu, session)
session.check_progress(now=sent + ProjectionSession.ACK_TIMEOUT + 0.5) # the waiting ACK is read, not timed out
assert session.acked == 2 and not session.pending
end_session(session, hu)
def test_car_that_stops_acking_times_out_at_the_new_limit(identity):
hu = FakeHeadUnit(identity)
session = streaming_session(hu, identity)
hu.hold_acks.set()
sent = sent_frame(hu, session)
session.check_progress(now=sent + 1.6) # past the old 1.5 s limit: no longer fatal
with pytest.raises(TimeoutError, match="2.5 s"):
session.check_progress(now=sent + ProjectionSession.ACK_TIMEOUT + 0.05)
end_session(session, hu)
def test_session_accepts_a_large_video_window_and_caps_it(identity):
hu = FakeHeadUnit(identity, window=100) # VW MIB3 offers 100
session = streaming_session(hu, identity)
assert session.window == ProjectionSession.MAX_WINDOW
end_session(session, hu)
def test_session_authentication_rejected(identity):
hu = FakeHeadUnit(identity, reject_auth=True)
session = connect(hu, identity)
@@ -797,6 +982,22 @@ def test_network_lease_retries_without_bssid_and_restores_previous():
assert nm.calls[-1] == "ActivateConnection" # home Wi-Fi comes back
def test_network_lease_waits_for_a_slow_honda_access_point(monkeypatch):
from openpilot.starpilot.system.android_auto import network
lease = network.NetworkLease(lambda *a, **k: None)
budgets = []
def attempt(credentials, timeout, cancelled):
budgets.append((credentials.bssid, timeout))
raise network.NetworkError("not up yet")
monkeypatch.setattr(lease, "_acquire_once", attempt)
monkeypatch.setattr(lease, "_drop_attempt", lambda: None)
with pytest.raises(network.NetworkError):
lease.acquire(credentials())
assert budgets == [("AA:BB:CC:DD:EE:FF", 90.0), ("", 30.0)]
def test_network_lease_does_not_undo_user_network_change():
nm = FakeNetworkManager()
lease = make_lease(nm)
@@ -237,3 +237,80 @@ def test_config_sanitizes_channel_cache(tmp_path):
config = identity_store.load_config(path)
assert config["rfcomm_cache"] == {CAR: 8} and config["auto_connect"] is True
assert identity_store.load_config(tmp_path / "missing.json")["rfcomm_cache"] is not identity_store.DEFAULT_CONFIG["rfcomm_cache"]
def test_config_v2_rgba_pipeline_upgrades_to_gpu_nv12(tmp_path):
from openpilot.starpilot.system.android_auto import identity as identity_store
path = tmp_path / "config.json"
# Every v2 save wrote these off; the car view could not do anything else then.
path.write_text(json.dumps({"config_version": 2, "gpu_nv12": False, "async_readback": False, "fps": 20}))
config = identity_store.load_config(path)
assert config["gpu_nv12"] is True and config["async_readback"] is True and config["fps"] == 20
identity_store.save_config(config, path)
# From v3 on they are a real choice and stay as saved.
path.write_text(json.dumps({**json.loads(path.read_text()), "gpu_nv12": False}))
assert identity_store.load_config(path)["gpu_nv12"] is False
def _wireless_backoff_harness(sup, monkeypatch, failure):
from openpilot.starpilot.system.android_auto import supervisor
now = [100.0]
monkeypatch.setattr(supervisor.time, "monotonic", lambda: now[0])
monkeypatch.setattr(sup, "_lease", lambda: supervisor.NoLease())
sup.config["connection"] = "wireless"
starts, waits = [], []
def attempt(_):
starts.append(now[0])
if len(starts) == 2:
sup._stop.set()
raise failure
def wait(seconds):
waits.append(seconds)
now[0] += seconds
if len(starts) == 1 and now[0] - starts[0] >= 5 and not sup._hfp_link.is_set():
sup._hfp_connected(CAR.lower()) # the car reaches out mid-backoff, e.g. the driver taps Android Auto
if sup._stop.is_set():
raise supervisor.Cancelled()
monkeypatch.setattr(sup, "_attempt", attempt)
monkeypatch.setattr(sup, "_wait", wait)
return starts, waits
def test_car_opening_hands_free_ends_the_backoff_early(identity, tmp_path, monkeypatch):
from openpilot.starpilot.system.android_auto import supervisor
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
sup.select_receiver(CAR, "Civic")
monkeypatch.setattr(supervisor, "BACKOFF_SECONDS", (30.0,))
starts, _ = _wireless_backoff_harness(sup, monkeypatch, RuntimeError("car not answering"))
sup._run(1)
assert len(starts) == 2
assert starts[1] - starts[0] < 6 # retried when the car connected, not after the 30 s backoff
def test_one_hands_free_connection_ends_only_one_backoff(identity, tmp_path, monkeypatch):
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
sup.select_receiver(CAR, "Civic")
sup._hfp_connected(CAR)
started = time.monotonic()
sup._wait_backoff(0.3, car_can_wake=True)
assert time.monotonic() - started < 0.1 # the car reached out: retry now
started = time.monotonic()
sup._wait_backoff(0.3, car_can_wake=True) # the link is still fresh, but that connection was used
assert time.monotonic() - started >= 0.25
sup._hfp_connected(CAR) # a new connection from the car ends the next one early again
started = time.monotonic()
sup._wait_backoff(0.3, car_can_wake=True)
assert time.monotonic() - started < 0.1
def test_car_ending_projection_still_gets_its_full_pause(identity, tmp_path, monkeypatch):
from openpilot.starpilot.system.android_auto import supervisor
from openpilot.starpilot.system.android_auto.session import PeerRequestedStop
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
sup.select_receiver(CAR, "Civic")
starts, waits = _wireless_backoff_harness(sup, monkeypatch, PeerRequestedStop("Head unit ended projection"))
sup._run(1)
assert waits[0] == supervisor.PEER_STOP_RETRY_SECONDS
@@ -0,0 +1,114 @@
from types import SimpleNamespace
from openpilot.starpilot.system.android_auto import current_car_ui
from openpilot.starpilot.system.android_auto.frame_source import FrameProducer, FrameRequest
STREAMS = SimpleNamespace(VISION_STREAM_NARROW_ROAD=0, VISION_STREAM_CABIN=1, VISION_STREAM_WIDE_ROAD=2)
STEP_MS = int(current_car_ui.CAMERA_WAIT_STEP * 1000)
class FakeSock:
"""A conflated CameraState socket and its poller: ``arrivals`` says whether each poll finds a message."""
def __init__(self):
self.arrivals: list[bool] = []
self.timeouts: list[int] = []
self._ready = False
def poll(self, timeout_ms: int):
self.timeouts.append(timeout_ms)
self._ready = self.arrivals.pop(0) if self.arrivals else False
return [self] if self._ready else []
def receive(self, non_blocking: bool = False):
ready, self._ready = self._ready, False
return b"msg" if ready else None
def make_pacer():
socks: dict[str, FakeSock] = {}
def factory(name):
socks[name] = FakeSock()
return socks[name], socks[name]
return current_car_ui.CameraPacer(sock_factory=factory, stream_types=STREAMS), socks
def test_draws_once_per_frame_of_the_shown_camera():
pacer, socks = make_pacer()
assert pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 10.0) # nothing seen yet: quiet, draw now
road = socks["narrowRoadCameraState"]
road.arrivals = [True, False, True, False]
assert pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 10.0)
assert not pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 10.01)
assert pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 10.05)
assert not pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 10.06) # same frame is never drawn twice
assert road.timeouts == [0, 0, STEP_MS, STEP_MS, STEP_MS]
def test_subscribes_only_to_the_camera_on_screen():
pacer, socks = make_pacer()
pacer.wait(STREAMS.VISION_STREAM_WIDE_ROAD, 10.0)
assert list(socks) == ["wideRoadCameraState"]
pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 10.0)
assert sorted(socks) == ["narrowRoadCameraState", "wideRoadCameraState"]
def test_quiet_camera_renders_at_the_encoder_rate():
pacer, socks = make_pacer()
pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 10.0)
road = socks["narrowRoadCameraState"]
road.arrivals = [True]
assert pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 10.0)
assert not pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 10.05)
# The camera stopped: every later frame is drawn without waiting, not one per CAMERA_MAX_GAP.
for i in range(10):
assert pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 10.001 + current_car_ui.CAMERA_MAX_GAP + i / 30)
assert road.timeouts[-10:] == [0] * 10
# And pacing resumes when it comes back.
road.arrivals = [True]
assert pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 11.0)
assert not pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 11.03)
def test_unknown_stream_does_not_block():
pacer, socks = make_pacer()
assert pacer.wait(99, 10.0)
assert socks == {}
def test_stream_names_match_the_services():
from openpilot.cereal.services import SERVICE_LIST
from openpilot.cereal.visionipc import VisionStreamType
pacer = current_car_ui.CameraPacer(sock_factory=lambda name: None)
assert {VisionStreamType(stream) for stream in pacer.state_for_stream} == \
{VisionStreamType.VISION_STREAM_NARROW_ROAD, VisionStreamType.VISION_STREAM_WIDE_ROAD, VisionStreamType.VISION_STREAM_CABIN}
for state in pacer.state_for_stream.values():
assert state in SERVICE_LIST
def test_twenty_hz_camera_inside_a_thirty_fps_budget_draws_every_camera_frame_once(tmp_path):
"""Simulate the renderer's gate: the 30 fps capture schedule, then the camera pacer."""
producer = FrameProducer(str(tmp_path / "frames"))
request = FrameRequest(1280, 720, 0, 0, 33_333)
pacer, socks = make_pacer()
camera_period, step = 0.05, 0.001
next_camera, drawn, frames = 10.0, [], 0
pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, 9.9)
road = socks["narrowRoadCameraState"]
now = 10.0
while now < 12.0:
if producer.capture_delay(request, int(now * 1e9)) == 0:
road.arrivals = [now >= next_camera]
if pacer.wait(STREAMS.VISION_STREAM_NARROW_ROAD, now):
if now >= next_camera:
frames += 1
next_camera += camera_period
drawn.append(now)
producer.advance(request, int(now * 1e9))
now = round(now + step, 6)
assert frames == 40 # every 20 Hz frame in two seconds
assert len(drawn) == frames # and nothing drawn in between
gaps = [b - a for a, b in zip(drawn, drawn[1:], strict=False)]
assert min(gaps) >= request.interval_us / 1e6 * 0.75 - 1e-9 # never faster than the encoder budget
@@ -4,7 +4,7 @@ import ast
from pathlib import Path
import unittest
from openpilot.starpilot.system.android_auto.current_car_ui import visible_geometry
from openpilot.starpilot.system.android_auto.current_car_ui import create_readback, finish_readback, visible_geometry
from openpilot.starpilot.system.android_auto.projection_geometry import projection_geometry
from openpilot.starpilot.system.android_auto.frame_source import FrameRequest
from openpilot.starpilot.system.android_auto.identity import DEFAULT_CONFIG
@@ -16,9 +16,9 @@ SOURCE = Path(__file__).parents[1]
class TestCurrentDisplaySource(unittest.TestCase):
def test_defaults_match_current_renderer_capabilities(self):
self.assertEqual(DEFAULT_CONFIG['view'], 'car')
self.assertFalse(DEFAULT_CONFIG['gpu_nv12'])
self.assertFalse(DEFAULT_CONFIG['async_readback'])
self.assertFalse(DEFAULT_CONFIG['render_profile'])
self.assertTrue(DEFAULT_CONFIG['gpu_nv12']) # current_car_ui converts with gpu_nv12.Nv12Converter
self.assertTrue(DEFAULT_CONFIG['async_readback']) # and publishes the asynchronous readback one step later
self.assertTrue(DEFAULT_CONFIG['render_profile']) # render_profile.RenderSampler, started by current_car_ui
def test_visible_geometry_never_crops_current_ui(self):
for width, height, margin_w, margin_h in ((1280, 720, 0, 0), (1920, 1080, 160, 80), (1280, 720, 0, 240), (800, 600, 0, 0)):
@@ -42,7 +42,7 @@ class TestCurrentDisplaySource(unittest.TestCase):
self.assertNotIn("openpilot.starpilot.ui.runtime_app", imports)
self.assertIn("openpilot.starpilot.system.android_auto.frame_source", imports)
self.assertNotIn("openpilot.starpilot.system.android_auto.ui.main", imports)
self.assertIn("producer.publish(request, readback.finish(), captured_ns)", source)
self.assertIn("producer.publish(request, pixels, in_flight_ns, pixel_format, advance=False)", source)
def test_view_uses_current_renderer(self):
source = (SOURCE / "view.py").read_text()
@@ -52,3 +52,36 @@ class TestCurrentDisplaySource(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestReadbackFallback(unittest.TestCase):
def test_async_failure_drops_frame_and_uses_same_size_sync_once(self):
from unittest.mock import Mock
factory = Mock()
synchronous = Mock(size=12, asynchronous=False)
synchronous.finish.return_value = memoryview(b"n" * 12)
factory.return_value = synchronous
asynchronous = Mock(size=12, asynchronous=True)
asynchronous.finish.side_effect = RuntimeError("GPU fence failed")
replacement, pixels = finish_readback(factory, asynchronous)
self.assertIsNone(pixels)
self.assertIs(replacement, synchronous)
asynchronous.close.assert_called_once()
factory.assert_called_once_with(12, asynchronous=False)
replacement, pixels = finish_readback(factory, replacement)
self.assertEqual(bytes(pixels), b"n" * 12)
factory.assert_called_once()
synchronous.finish.side_effect = RuntimeError("synchronous failure")
with self.assertRaises(RuntimeError):
finish_readback(factory, synchronous)
def test_async_initialization_falls_back_without_changing_explicit_sync(self):
from unittest.mock import Mock, call
synchronous = Mock(asynchronous=False)
factory = Mock(side_effect=[RuntimeError("PBO allocation failed"), synchronous])
self.assertIs(create_readback(factory, 12, True), synchronous)
self.assertEqual(factory.call_args_list, [call(12, asynchronous=True), call(12, asynchronous=False)])
factory = Mock(side_effect=RuntimeError("readback unavailable"))
with self.assertRaises(RuntimeError):
create_readback(factory, 12, False)
factory.assert_called_once_with(12, asynchronous=False)
@@ -0,0 +1,439 @@
"""On-device Android Auto diagnostics: richer session logs, readable reports and the renderer profile."""
from collections import deque
import io
import json
import socket
import ssl
import sys
import time
import zipfile
from types import SimpleNamespace
import pytest
from openpilot.starpilot.system.android_auto import bootstrap as bs
from openpilot.starpilot.system.android_auto import compat_report
from openpilot.starpilot.system.android_auto.render_profile import RenderSampler, RenderSummary
from openpilot.starpilot.system.android_auto.session import ProjectionSession, choose_video_mode, redact_head_unit
from openpilot.starpilot.system.android_auto.tests.fake_head_unit import FakeHeadUnit, make_identity
from openpilot.starpilot.system.android_auto.wire import DESCRIBE_TEXT, describe, field, parse_fields
@pytest.fixture(scope="module")
def identity(tmp_path_factory):
return make_identity(tmp_path_factory.mktemp("identity"))
def connect(hu: FakeHeadUnit, identity, log, ca=None) -> ProjectionSession:
sock = socket.create_connection(("127.0.0.1", hu.port), timeout=5)
return ProjectionSession(sock, str(identity["phone_cert"]), str(identity["phone_key"]), log,
str(ca or identity["root"]))
# ------------------------------------------------------------- session logs
def test_describe_keeps_text_and_bounds_opaque_bytes():
described = describe(field(1, "Honda") + field(2, 7) + field(3, field(1, "ALPSALPINE")) + field(4, b"\xff" * 100))
assert described[1] == ["Honda"] and described[2] == [7] and described[3] == [{1: ["ALPSALPINE"]}]
assert described[4] == [{"hex": "ff" * 64, "bytes": 100}]
assert describe(field(1, "x" * (DESCRIBE_TEXT + 1)))[1] == ["x" * DESCRIBE_TEXT + "…"]
def test_ignored_messages_are_rate_limited():
counts = []
session = ProjectionSession.__new__(ProjectionSession)
session._log = lambda name, **values: counts.append(values["count"])
session.ignored_counts = {}
for _ in range(250):
session.ignored("channel_ignored", 9, 0x8001, b"\x08\x01")
assert counts == [1, 2, 3, 4, 5, 100, 200]
def test_no_supported_mode_error_lists_the_offer():
channels = [{"id": 3, "video_configs": [parse_fields(field(1, 4) + field(10, 3)), parse_fields(field(1, 2) + field(10, 7))]},
{"id": 5, "display_type": 1, "video_configs": [parse_fields(field(1, 1) + field(10, 5))]}]
with pytest.raises(ValueError, match=r"2560x1440 H\.264, 1280x720 H\.265, 800x480 VP9 \(display 1\)"):
choose_video_mode(channels)
def test_discovery_log_describes_the_car_and_its_channels(identity):
records = []
hu = FakeHeadUnit(identity)
session = connect(hu, identity, lambda name, **values: records.append((name, values)))
session.authenticate()
session.start("StarPilot", "comma.ai")
discovered = next(values for name, values in records if name == "discovered")
assert discovered["head_unit"][2] == ["Honda"] and discovered["head_unit"][3] == ["Civic"]
video = next(channel for channel in discovered["channels"] if channel.get("video_configs"))
assert video["descriptor"] and "display_type" in video
session.shutdown()
session.peer.close()
hu.thread.join(5)
def test_vehicle_ids_are_redacted_from_discovery_and_version_logs():
described = redact_head_unit({2: ["Honda"], 5: ["VIN-SECRET"], 17: [{1: ["Honda"], 4: ["VIN-SECRET"]}]})
assert described[2] == ["Honda"] and "VIN-SECRET" not in json.dumps(described)
version = bs.describe_version_request(field(1, 4) + field(2, 1) + field(5, field(1, "Honda") + field(4, "VIN-SECRET")) +
field(4, field(1, "192.168.1.1") + field(2, 5288) + field(4, 36)))
assert version[5][0] == {1: ["Honda"], 4: ["redacted"]}
assert version[4][0][4] == [36], "a channel number in the endpoint is kept"
def test_tls_failure_is_logged_with_its_reason(identity, tmp_path):
records = []
hu = FakeHeadUnit(identity)
other_root = make_identity(tmp_path)["root"] # the car's certificate does not chain to this root
session = connect(hu, identity, lambda name, **values: records.append((name, values)), ca=other_root)
with pytest.raises(ssl.SSLError):
session.authenticate()
failed = next(values for name, values in records if name == "tls_failed")
assert failed["reason"] and failed["error"]
session.peer.close()
hu.close()
def ack_session():
session = ProjectionSession.__new__(ProjectionSession)
session.retired_sessions = deque()
session.session_id = 1
session.config_ack_slack = 1
session.unacked = 1
session.pending = deque([time.monotonic()])
session.max_ack_seconds = 0
session.acked = session.epoch_acked = 0
session.video_confirmed = False
events = []
session._log = lambda name, **values: events.append(name)
return session, events
def test_configuration_ack_alone_does_not_confirm_video():
session, events = ack_session()
session._handle_ack(1, 1)
assert not events
assert session.unacked == 0 # preserve the existing flow-control behavior
session._handle_ack(1, 1) # the additional ACK proves at least one was a frame
assert events == ["video_acknowledged"]
@pytest.mark.parametrize("separate_config", [False, True])
def test_video_confirmation_with_optional_or_separate_config_ack(separate_config):
session, events = ack_session()
if separate_config:
session._handle_ack(0, 1)
assert not events
session._handle_ack(1, 1)
if not separate_config:
assert not events # one ACK is still ambiguous
session.unacked = 1
session.pending.append(time.monotonic())
session._handle_ack(1, 1)
assert events == ["video_acknowledged"]
def test_combined_config_and_frame_ack_confirms_video():
session, events = ack_session()
session._handle_ack(1, 2)
assert events == ["video_acknowledged"]
# ------------------------------------------------------------------ reports
# Shaped like the Civic's session 25 (2026-09-28): a refused start, then a projection that later dropped.
CIVIC_HEAD_UNIT = {"car_make": "Honda", "car_model": "T20", "car_year": "2025", "head_unit_make": "ALPSALPINE",
"head_unit_model": "8A501-T20-A1"}
CIVIC_SESSION = [
{"t": "2026-09-28T21:13:51+00:00", "event": "session_start", "receiver": "Honda CIVIC", "trigger": "onroad"},
{"event": "stage", "state": "connecting_bluetooth"},
{"event": "stage", "state": "rfcomm"},
{"event": "bootstrap_version", "major": 4, "minor": 1, "head_unit": CIVIC_HEAD_UNIT},
{"event": "bootstrap_start_refused", "status": -1, "endpoint": False},
{"event": "bootstrap_credentials", "ssid": "Car-4E91", "security": "wpa2/wpa3", "ap_type": 0, "key_length": 12},
{"event": "stage", "state": "joining_wifi"},
{"event": "wifi_joined", "interface": "wlan0", "local_ip": "192.168.180.183"},
{"event": "stage", "state": "authenticating"},
{"event": "version", "major": 4, "minor": 1, "reply": "6.1"},
{"event": "tls_established", "version": "TLSv1.2", "cipher": "ECDHE-RSA-AES256-GCM-SHA384"},
{"event": "discovered", "head_unit": {"2": ["Honda"], "3": ["Civic"], "5": ["redacted"]},
"channels": [{"id": 22, "services": [3], "display_type": 0, "video_configs": [{"1": [2], "10": [3]}, {"1": [1], "10": [3]}]}]},
{"event": "projection_ready", "mode": {"width": 1280, "height": 720, "fps": 60, "margin_width": 0, "margin_height": 0}},
{"event": "stage", "state": "streaming"},
{"event": "video_focus", "focused": True},
{"event": "video_acknowledged", "session": 1, "ack_ms": 29},
{"event": "channel_ignored", "channel": 9, "kind": 32769, "count": 1, "message": {"1": [1]}},
{"event": "attempt_failed", "stage": "streaming", "error": "projecting: Video acknowledgement older than 1.5 s"},
]
def test_report_summarizes_a_civic_session():
report = compat_report.summarize(CIVIC_SESSION)
assert report["transport"] == "wireless" and report["trigger"] == "onroad"
assert report["car"]["head_unit_make"] == "ALPSALPINE" and report["car"]["car_model"] == "Civic"
assert report["furthest_stage"] == "streaming" and report["wifi"]["security"] == "wpa2/wpa3"
assert report["video"]["offered"] == ["1280x720 H.264", "800x480 H.264"]
assert report["outcome"] == "projected, then failed: projecting: Video acknowledgement older than 1.5 s"
text = compat_report.render_text(report, "session-000025.jsonl")
assert "Outcome: projected, then failed" in text and "cipher: ECDHE-RSA-AES256-GCM-SHA384" in text
assert "Not handled: channel_ignored channel 9" in text
assert "Car-4E91" not in text and "192.168.180.183" not in text # no Wi-Fi name or address in a shared report
def test_successful_retry_clears_outcome_failure_but_keeps_history():
failure = {"event": "attempt_failed", "stage": "streaming", "error": "link lost"}
events = [{"event": "video_acknowledged"}, failure]
assert compat_report.summarize(events)["outcome"] == "projected, then failed: link lost"
events.append({"event": "video_acknowledged"})
report = compat_report.summarize(events)
assert report["outcome"] == "projected"
assert report["errors"] == [{"stage": "streaming", "error": "link lost"}]
def write_log(directory, name, events):
path = directory / name
path.write_text("".join(json.dumps(event) + "\n" for event in events) + '{"cut short')
return path
def test_session_logs_are_newest_first_by_number(tmp_path):
write_log(tmp_path, "session-000009-20260927-121412.jsonl", [])
write_log(tmp_path, "session-000010-20260728-080505.jsonl", []) # the clock had not synced yet
write_log(tmp_path, "session-20260101-000000.jsonl", [])
(tmp_path / "notes.txt").write_text("")
assert [path.name[:14] for path in compat_report.session_logs(tmp_path)] == \
["session-000010", "session-000009", "session-202601"]
assert compat_report.session_path("../config.json", tmp_path) is None
def test_bundle_has_logs_reports_and_settings_without_addresses(tmp_path):
logs = tmp_path / "logs"
logs.mkdir()
write_log(logs, "session-000025-20260928-141351.jsonl", CIVIC_SESSION)
(logs / "render_profile.txt").write_text("==== profile ====\n")
config = tmp_path / "config.json"
config.write_text(json.dumps({"config_version": 2, "receiver_address": "AA:BB:CC:DD:EE:FF", "receiver_name": "Honda CIVIC",
"rfcomm_cache": {"AA:BB:CC:DD:EE:FF": 5}}))
with zipfile.ZipFile(io.BytesIO(compat_report.bundle(logs, config))) as archive:
names = set(archive.namelist())
assert {"REPORT.txt", "config.json", "logs/session-000025-20260928-141351.jsonl",
"reports/session-000025-20260928-141351.json", "logs/render_profile.txt"} <= names
shared = archive.read("config.json").decode()
assert "AA:BB:CC:DD:EE:FF" not in shared and json.loads(shared)["rfcomm_cache"] == [5]
assert "projected, then failed" in archive.read("REPORT.txt").decode()
assert not any(name.endswith((".pem", ".key")) for name in names)
def test_report_tool_prints_a_log(tmp_path, monkeypatch, capsys):
from openpilot.tools.android_auto import compat_report as tool
path = write_log(tmp_path, "session-000025-20260928-141351.jsonl", CIVIC_SESSION)
monkeypatch.setattr(sys, "argv", ["compat_report.py", str(path)])
assert tool.main() == 0
assert "Outcome: projected, then failed" in capsys.readouterr().out
monkeypatch.setattr(sys, "argv", ["compat_report.py", str(path), "--json"])
assert tool.main() == 0
assert json.loads(capsys.readouterr().out)["car"]["head_unit_make"] == "ALPSALPINE"
# ----------------------------------------------------------- render profile
def hot_render_function(sampler):
sampler.sample(sys._getframe())
def test_render_sampler_counts_render_stacks_and_idle(tmp_path):
clock = [0.0]
sampler = RenderSampler(tmp_path / "render_profile.txt", clock=lambda: clock[0])
sampler.sample(sys._getframe()) # renderer idle
sampler.rendering, sampler.onroad = True, True
for _ in range(3):
hot_render_function(sampler)
sampler.summary = {"event": "render_stats", "fps": 29.5, "frame_ms": 21.0}
clock[0] = 60.0
text = sampler.report()
assert "onroad (100% onroad), rendering 75% of 4 samples" in text
assert "render_stats fps=29.5 frame_ms=21.0" in text
innermost = text.split("-- innermost function")[1].split("-- innermost line")[0]
assert "100.0% test_diagnostics_logging.py" in innermost and "hot_render_function" in innermost
sampler.flush()
assert (tmp_path / "render_profile.txt").read_text() == text
assert sampler.samples == 0 and sampler.report() == ""
def test_sampler_reports_parameter_keys_without_values(tmp_path):
sampler = RenderSampler(tmp_path / "profile.txt")
sampler.rendering = True
class Code:
co_name = "get"
co_filename = "/data/openpilot/openpilot/common/params.py"
co_firstlineno = 220
sampler.sample(SimpleNamespace(f_code=Code(), f_lineno=222, f_locals={"key": "BorderWidth", "value": "private-value"}, f_back=None))
report = sampler.report()
assert "100.0% get(BorderWidth)" in report and "private-value" not in report
def test_render_sampler_file_rotates_at_its_size_cap(tmp_path):
path = tmp_path / "render_profile.txt"
sampler = RenderSampler(path, max_bytes=300)
for index in range(5):
sampler._write(f"window {index} " + "x" * 100 + "\n")
assert path.read_text().startswith("window 4")
assert (tmp_path / "render_profile.1.txt").read_text().startswith("window 2")
assert path.stat().st_size <= 300 and not (tmp_path / "render_profile.2.txt").exists()
def test_render_sampler_thread_samples_the_render_thread(tmp_path):
sampler = RenderSampler(tmp_path / "render_profile.txt", interval=0.001, window=3600)
sampler.rendering = True
sampler.start()
deadline = time.monotonic() + 2.0
while sampler.samples < 5 and time.monotonic() < deadline:
sum(range(10_000))
sampler.close()
assert "test_render_sampler_thread_samples_the_render_thread" in (tmp_path / "render_profile.txt").read_text()
def test_render_summary_reports_frame_rate_and_time_per_window():
summary = RenderSummary(0.0, interval=1.0)
for began in (0.0, 0.25, 0.5, 0.75):
assert summary.frame_done(began, began + 0.05) is None
assert summary.frame_done(0.95, 1.0) == {"event": "render_stats", "fps": 5.0, "frame_ms": 50.0}
assert summary.frames == 0 and summary.started == 1.0 # the next window starts where this one ended
# ------------------------------------------------------------- session log writer
def _read_records(directory):
(path,) = directory.glob("session-*.jsonl")
return [json.loads(line) for line in path.read_text().splitlines()]
def test_session_log_writes_in_order_off_the_caller_thread(tmp_path):
from openpilot.starpilot.system.android_auto.supervisor import EventLog
log = EventLog(tmp_path)
log.open()
for index in range(50):
log("stats", index=index)
log.close()
records = _read_records(tmp_path)
assert [record["index"] for record in records] == list(range(50))
assert len(log.recent) == 40 and log.recent[-1]["index"] == 49
def test_session_log_never_blocks_on_a_stalled_disk(tmp_path, monkeypatch):
import threading
import time
from openpilot.starpilot.system.android_auto import supervisor
monkeypatch.setattr(supervisor, "LOG_QUEUE_MAX", 5)
release = threading.Event()
original = supervisor.EventLog._write
def stalled(handle, records, stopped):
release.wait(5) # the disk is stuck until the test lets it go
original(handle, records, stopped)
monkeypatch.setattr(supervisor.EventLog, "_write", staticmethod(stalled))
log = supervisor.EventLog(tmp_path)
log.open()
started = time.monotonic()
for index in range(20):
log("stats", index=index)
assert time.monotonic() - started < 0.5 # the caller never waited for the disk
release.set()
deadline = time.monotonic() + 2
while not log._queue.empty() and time.monotonic() < deadline:
time.sleep(0.01) # the disk catches up
log("after", index=20)
log.close()
records = _read_records(tmp_path)
dropped = [record for record in records if record["event"] == "log_dropped"]
assert dropped and dropped[0]["count"] == 15
assert [record["index"] for record in records if record["event"] != "log_dropped"] == [0, 1, 2, 3, 4, 20]
def test_session_log_without_a_file_keeps_the_recent_tail(tmp_path):
from openpilot.starpilot.system.android_auto.supervisor import EventLog
(tmp_path / "file").write_text("")
log = EventLog(tmp_path / "file" / "logs") # cannot be created: a file is in the way
log.open()
log("stage", stage="rfcomm")
log.close()
assert log.recent[-1]["event"] == "stage"
def test_log_reads_reject_replaced_symlinks_and_directories(tmp_path, monkeypatch):
logs = tmp_path / "logs"
logs.mkdir()
outside = tmp_path / "private"
outside.write_text("private material")
session = logs / "session-000001-test.jsonl"
session.symlink_to(outside)
assert compat_report.session_logs(logs) == []
with pytest.raises(OSError):
compat_report.read_log(session, compat_report.MAX_LOG_BYTES)
# A path substituted after listing must not bypass the opened-file checks.
monkeypatch.setattr(compat_report, "session_logs", lambda *_: [session])
with pytest.raises(OSError):
compat_report.bundle(logs)
session.unlink()
session.mkdir()
with pytest.raises(OSError):
compat_report.read_log(session, compat_report.MAX_LOG_BYTES)
def test_log_and_bundle_reads_enforce_size_budgets(tmp_path, monkeypatch):
session = tmp_path / "session-000001-test.jsonl"
session.write_bytes(b"x" * 9)
with pytest.raises(OSError):
compat_report.read_log(session, 8)
assert compat_report.read_log(session, 8, tail=True) == b"x" * 8
data = (json.dumps({"event": "session_stop"}) + "\n").encode() * 400
session.write_bytes(data)
(tmp_path / "session-000002-test.jsonl").write_bytes(data)
monkeypatch.setattr(compat_report, "MAX_BUNDLE_BYTES", len(data) + 4096)
with pytest.raises(OSError):
compat_report.bundle(tmp_path)
def test_session_log_full_queue_close_then_reopen_retires_old_writer(tmp_path, monkeypatch):
import threading
from openpilot.starpilot.system.android_auto import supervisor
monkeypatch.setattr(supervisor, "LOG_QUEUE_MAX", 2)
release = threading.Event()
entered = threading.Event()
original = supervisor.EventLog._write
handles = []
def stalled_once(handle, records, stopped):
handles.append(handle)
if len(handles) == 1:
entered.set()
release.wait(5)
original(handle, records, stopped)
monkeypatch.setattr(supervisor.EventLog, "_write", staticmethod(stalled_once))
log = supervisor.EventLog(tmp_path)
log.open()
old_writer = log._writer
assert entered.wait(1)
try:
log("first", index=0)
log("first", index=1)
assert log._queue.full()
log.close()
log.open()
log("second", index=2)
log.close()
finally:
release.set()
old_writer.join(timeout=2)
log.close()
assert not old_writer.is_alive()
assert all(handle.closed for handle in handles)
files = sorted(tmp_path.glob("session-*.jsonl"))
first = [json.loads(line) for line in files[0].read_text().splitlines()]
second = [json.loads(line) for line in files[1].read_text().splitlines()]
assert [item["event"] for item in first] == ["first", "first"]
assert [item["event"] for item in second] == ["second"]
@@ -0,0 +1,99 @@
"""Pixel-level GPU regression checks; run with STARPILOT_AUTO_GL_TEST=1 on a desktop or comma.
The expected image is the car view's RGBA composition read back, followed by the
CPU NV12 reference. This catches flipped rows, margin placement and chroma blocks
that straddle the padding without duplicating the shader's coordinate logic.
"""
import ctypes as C
import os
import sys
from unittest import SkipTest
import numpy as np
import pytest
from openpilot.starpilot.system.android_auto import gpu_nv12, headless_egl
@pytest.mark.parametrize("margins", [(-1, 0), (32, 0), (0, -1), (0, 18)])
def test_invalid_margins_leave_no_gpu_resources(margins):
with pytest.raises(ValueError, match="margins"):
gpu_nv12.Nv12Converter(32, 18, margin_w=margins[0], margin_h=margins[1])
@pytest.fixture(scope="module")
def gpu():
if os.getenv("STARPILOT_AUTO_GL_TEST") != "1":
raise SkipTest("STARPILOT_AUTO_GL_TEST=1 enables tests requiring a real graphics context")
import pyray as rl
with pytest.MonkeyPatch.context() as patch:
if sys.platform == "darwin":
gl = C.CDLL("/System/Library/Frameworks/OpenGL.framework/OpenGL")
# A windowless CGL context also works in a terminal without a Cocoa app.
gl.CGLChoosePixelFormat.argtypes = [C.POINTER(C.c_int), C.POINTER(C.c_void_p), C.POINTER(C.c_int)]
gl.CGLCreateContext.argtypes = [C.c_void_p, C.c_void_p, C.POINTER(C.c_void_p)]
gl.CGLSetCurrentContext.argtypes = [C.c_void_p]
gl.CGLDestroyContext.argtypes = [C.c_void_p]
gl.CGLDestroyPixelFormat.argtypes = [C.c_void_p]
pixel_format, context, count = C.c_void_p(), C.c_void_p(), C.c_int()
# kCGLPFAOpenGLProfile, kCGLOGLPVersion_3_2_Core, kCGLPFAColorSize
assert gl.CGLChoosePixelFormat((C.c_int * 5)(99, 0x3200, 8, 24, 0), C.byref(pixel_format), C.byref(count)) == 0
assert gl.CGLCreateContext(pixel_format, None, C.byref(context)) == 0
gl.CGLDestroyPixelFormat(pixel_format)
assert gl.CGLSetCurrentContext(context) == 0
patch.setattr(headless_egl.C, "CDLL", lambda _: gl)
loader_type = C.CFUNCTYPE(C.c_void_p, C.c_char_p)
def address(name):
function = getattr(gl, name.decode(), None)
return C.cast(function, C.c_void_p).value if function is not None else None
loader = loader_type(address)
rl.rl_load_extensions(rl.ffi.cast("void *", C.cast(loader, C.c_void_p).value))
rl.rlgl_init(64, 64)
rl.rl_set_framebuffer_width(64)
rl.rl_set_framebuffer_height(64)
def close():
rl.rlgl_close()
gl.CGLSetCurrentContext(None)
gl.CGLDestroyContext(context)
else:
context = headless_egl.HeadlessContext(64, 64)
close = context.close
try:
yield rl
finally:
close()
def read_frame(size, regions, asynchronous=False):
readback = headless_egl.FrameReadback(size, asynchronous=asynchronous)
try:
readback.start(regions)
return bytes(readback.finish())
finally:
readback.close()
@pytest.mark.parametrize("width,height,margin_w,margin_h", [(32, 18, 0, 0), (32, 18, 4, 2), (32, 20, 3, 5), (40, 18, 0, 6)])
@pytest.mark.parametrize("asynchronous", [False, True])
def test_composed_car_frame_converts_to_the_same_picture(gpu, width, height, margin_w, margin_h, asynchronous):
"""current_car_ui composes (fit, margins, flip) into the output target, then converts that target as is."""
rl = gpu
logical_w, logical_h = 31, 18 # the projection's logical canvas, scaled to fit the visible area
pixels = np.random.default_rng(11).integers(0, 256, (logical_h, logical_w, 4), dtype=np.uint8)
buffer = rl.ffi.from_buffer(pixels)
texture_id = rl.rl_load_texture(rl.ffi.cast("void *", buffer), logical_w, logical_h, 7, 1)
source = rl.Texture(texture_id, logical_w, logical_h, 1, 7)
output = rl.load_render_texture(width, height)
converter = gpu_nv12.Nv12Converter(width, height)
try:
gpu_nv12.compose_rgba(source, output, margin_w, margin_h, fit=True)
rgba = read_frame(width * height * 4, [(output.id, width, height, 0)])
nv12 = read_frame(width * height * 3 // 2, converter.convert(output.texture), asynchronous)
assert nv12 == gpu_nv12.reference_nv12(rgba, width, height)
finally:
converter.close()
rl.unload_render_texture(output)
rl.rl_unload_texture(texture_id)
@@ -98,3 +98,47 @@ def json_fields(fields: dict) -> dict:
"""Keep unknown protobuf bytes inspectable without guessing their schema."""
return {number: [{"hex": value.hex()} if isinstance(value, bytes) else value for value in values]
for number, values in fields.items()}
DESCRIBE_TEXT = 512
DESCRIBE_DEPTH = 4
DESCRIBE_REPEATED = 32 # values kept per field; a head unit lists at most a few dozen keycodes or sensors
DESCRIBE_HEX = 64 # bytes of an opaque value kept as hex
def _describe_value(value, depth: int):
if not isinstance(value, bytes):
return value
try:
decoded = value.decode("utf-8")
if decoded and decoded.isprintable():
return decoded[:DESCRIBE_TEXT] if len(decoded) <= DESCRIBE_TEXT else decoded[:DESCRIBE_TEXT] + "…"
except UnicodeDecodeError:
pass
if depth > 0 and value:
try:
return describe_fields(parse_fields(value), depth - 1)
except ValueError:
pass
described = {"hex": value[:DESCRIBE_HEX].hex()}
if len(value) > DESCRIBE_HEX:
described["bytes"] = len(value)
return described
def describe_fields(fields: dict, depth: int = DESCRIBE_DEPTH) -> dict:
"""A bounded, readable view of a message whose schema we may not know, for diagnostics.
Text stays text and nested messages are decoded, so a head unit's make, model and
capabilities read directly in a log; anything else is a short hex prefix.
"""
return {number: [_describe_value(value, depth) for value in values[:DESCRIBE_REPEATED]]
for number, values in fields.items()}
def describe(data: bytes, depth: int = DESCRIBE_DEPTH):
"""``describe_fields`` of raw bytes, or their hex prefix when they are not protobuf."""
try:
return describe_fields(parse_fields(data), depth)
except ValueError:
return _describe_value(data, 0)
@@ -0,0 +1,46 @@
#!/usr/bin/env python3
"""Summarize Android Auto session logs into a shareable compatibility report.
On the comma, with no arguments it reports the newest session:
cd /data/openpilot
PYTHONPATH=/data/openpilot /usr/local/venv/bin/python openpilot/tools/android_auto/compat_report.py
... compat_report.py --all # every kept session, newest first
... compat_report.py session-000012-*.jsonl --json
... compat_report.py --bundle /tmp/android_auto.zip # logs, reports and settings in one zip
Logs pulled to another machine work too: pass their paths.
"""
import argparse
import json
import sys
from pathlib import Path
from openpilot.starpilot.system.android_auto import compat_report
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("logs", nargs="*", type=Path, help="session logs (default: the newest on this comma)")
parser.add_argument("--all", action="store_true", help="every kept session on this comma")
parser.add_argument("--json", action="store_true", help="print the report as JSON")
parser.add_argument("--bundle", type=Path, help="write a zip of every log, report and the settings here")
args = parser.parse_args()
if args.bundle:
args.bundle.write_bytes(compat_report.bundle())
print(f"wrote {args.bundle}")
return 0
logs = args.logs or (compat_report.session_logs() if args.all else compat_report.session_logs()[:1])
if not logs:
print("No Android Auto session logs found.", file=sys.stderr)
return 1
for path in logs:
report = compat_report.summarize(compat_report.load_events(path))
print(json.dumps({"log": path.name, **report}, indent=2, default=str) if args.json else compat_report.render_text(report, path.name))
return 0
if __name__ == "__main__":
raise SystemExit(main())