Big Tooth & Demos

This commit is contained in:
firestarsdog
2026-08-30 03:30:02 -04:00
parent 6420724913
commit 16e561c5b5
7 changed files with 808 additions and 17 deletions
+23
View File
@@ -607,6 +607,27 @@ launch_python_ui() {
UI_PIDS+=("$!")
}
configure_bluetooth_demo() {
case " ${UI_TARGETS[*]-} " in
*" c3 "*|*" c4 "*)
local fake_bluetooth="${SP_ONROAD_FAKE_BLUETOOTH:-}"
if [[ -z "${fake_bluetooth}" ]]; then
case " ${UI_TARGETS[*]-} " in
" c3 ") fake_bluetooth="${SP_C3_FAKE_BLUETOOTH:-1}" ;;
" c4 ") fake_bluetooth="${SP_C4_FAKE_BLUETOOTH:-1}" ;;
*) fake_bluetooth=1 ;;
esac
fi
if env_var_truthy "${fake_bluetooth}"; then
export SP_ALLOW_DESKTOP_FAKE_BLUETOOTH=1
else
export SP_ALLOW_DESKTOP_FAKE_BLUETOOTH=0
fi
;;
esac
}
launch_control_bar() {
local watch_pids="${UI_PIDS[*]}"
(
@@ -674,6 +695,8 @@ if [[ "${REPLAY_ONLY}" != "1" && ${#UI_TARGETS[@]} -eq 0 ]]; then
exit 1
fi
configure_bluetooth_demo
echo "Preparing replay and desktop UI runtime..."
build_replay
+5
View File
@@ -55,6 +55,11 @@ export SKIP_FW_QUERY=1
export USE_WEBCAM=1
export PRIME_TYPE="${PRIME_TYPE:-0}"
export SP_C3_FAKE_DRIVE_STATS="${SP_C3_FAKE_DRIVE_STATS:-1}"
if [[ "${SP_C3_FAKE_BLUETOOTH:-1}" =~ ^(1|true|yes|on)$ ]]; then
export SP_ALLOW_DESKTOP_FAKE_BLUETOOTH=1
else
export SP_ALLOW_DESKTOP_FAKE_BLUETOOTH=0
fi
backup_dir="$(mktemp -d /tmp/starpilot_c3_ui_backup.XXXXXX)"
backup_manifest="${backup_dir}/.artifact_manifest"
+9 -3
View File
@@ -8,10 +8,12 @@ from openpilot.selfdrive.ui.layouts.settings.starpilot.main_panel import StarPil
from openpilot.selfdrive.ui.layouts.settings.software import SoftwareLayout
from openpilot.selfdrive.ui.layouts.settings.toggles import TogglesLayout
from openpilot.system.ui.lib.application import gui_app, FontWeight, MousePos
from openpilot.system.ui.lib.bluetooth_manager import BluetoothManager
from openpilot.system.ui.lib.multilang import tr, tr_noop
from openpilot.system.ui.lib.text_measure import measure_text_cached
from openpilot.system.ui.lib.wifi_manager import WifiManager
from openpilot.system.ui.widgets import Widget
from openpilot.system.ui.widgets.bluetooth import BluetoothManagerUI
from openpilot.system.ui.widgets.network import NetworkUI
# Constants
@@ -38,9 +40,10 @@ class PanelType(IntEnum):
STARPILOT = 0
DEVICE = 1
NETWORK = 2
TOGGLES = 3
SOFTWARE = 4
DEVELOPER = 5
BLUETOOTH = 3
TOGGLES = 4
SOFTWARE = 5
DEVELOPER = 6
@dataclass
@@ -68,11 +71,14 @@ class SettingsLayout(Widget):
# Panel configuration
wifi_manager = WifiManager()
wifi_manager.set_active(False)
bluetooth_manager = BluetoothManager()
bluetooth_manager.set_active(False)
self._panels = {
PanelType.STARPILOT: PanelInfo(tr_noop("StarPilot"), StarPilotLayout()),
PanelType.DEVICE: PanelInfo(tr_noop("Device"), DeviceLayout()),
PanelType.NETWORK: PanelInfo(tr_noop("Network"), NetworkUI(wifi_manager)),
PanelType.BLUETOOTH: PanelInfo(tr_noop("Bluetooth"), BluetoothManagerUI(bluetooth_manager)),
PanelType.TOGGLES: PanelInfo(tr_noop("Toggles"), TogglesLayout()),
PanelType.SOFTWARE: PanelInfo(tr_noop("Software"), SoftwareLayout()),
PanelType.DEVELOPER: PanelInfo(tr_noop("Developer"), DeveloperLayout()),
+172
View File
@@ -0,0 +1,172 @@
import os
from types import SimpleNamespace
os.environ.setdefault("SP_HEADLESS_TEST", "1")
from openpilot.starpilot.system.bluetooth.protocol import BluetoothDevice, BluetoothStatus
import openpilot.selfdrive.ui.layouts.settings.settings as settings_module
from openpilot.selfdrive.ui.layouts.settings.settings import PanelType, SettingsLayout
from openpilot.system.ui.widgets import DialogResult
from openpilot.system.ui.widgets.bluetooth import (BluetoothManagerUI, PANEL_BACKGROUND, ROW_BORDER,
device_action_allowed, device_status_text)
ADDRESS = "00:11:22:33:44:55"
class FakeBluetoothManager:
def __init__(self, status: BluetoothStatus):
self.status = status
self.calls = []
def operation_for(self, _address: str) -> str:
return ""
def pair(self, address: str):
self.calls.append(("pair", address))
def connect(self, address: str):
self.calls.append(("connect", address))
def disconnect(self, address: str):
self.calls.append(("disconnect", address))
def set_scanning(self, enabled: bool):
self.calls.append(("scan", enabled))
def respond(self, prompt_id: str, accepted: bool, value: str = ""):
self.calls.append(("respond", prompt_id, accepted, value))
def make_ui(manager: FakeBluetoothManager) -> BluetoothManagerUI:
ui = object.__new__(BluetoothManagerUI)
ui._manager = manager
ui._scan_pending = False
return ui
def make_device(**overrides) -> BluetoothDevice:
return BluetoothDevice(ADDRESS, "Speaker", **overrides)
def test_bluetooth_is_a_standalone_settings_panel():
assert PanelType.BLUETOOTH.value == PanelType.NETWORK.value + 1
assert PanelType.TOGGLES.value == PanelType.BLUETOOTH.value + 1
def test_bluetooth_uses_the_network_panel_black_surface():
def rgba(color):
return tuple(color) if isinstance(color, tuple) else (color.r, color.g, color.b, color.a)
assert rgba(PANEL_BACKGROUND) == (0, 0, 0, 255)
assert rgba(ROW_BORDER) == (200, 200, 200, 255)
def test_settings_constructs_a_dedicated_bluetooth_panel(monkeypatch):
class PanelStub:
def __init__(self, *_args):
pass
def set_depth_callback(self, _callback):
pass
def set_settings_layout(self, _layout):
pass
class ManagerStub:
def __init__(self):
self.active = None
def set_active(self, active):
self.active = active
captured = {}
class BluetoothPanelStub(PanelStub):
def __init__(self, manager):
captured["manager"] = manager
for name in ("StarPilotLayout", "DeviceLayout", "TogglesLayout", "SoftwareLayout", "DeveloperLayout", "NetworkUI"):
monkeypatch.setattr(settings_module, name, PanelStub)
monkeypatch.setattr(settings_module, "WifiManager", ManagerStub)
monkeypatch.setattr(settings_module, "BluetoothManager", ManagerStub)
monkeypatch.setattr(settings_module, "BluetoothManagerUI", BluetoothPanelStub)
monkeypatch.setattr(settings_module.gui_app, "font", lambda *_args: object())
monkeypatch.setattr(settings_module.gui_app, "texture", lambda *_args: object())
layout = SettingsLayout()
assert PanelType.BLUETOOTH in layout._panels
assert isinstance(captured["manager"], ManagerStub)
assert captured["manager"].active is False
def test_device_status_prioritizes_operations_then_connection_and_capabilities():
device = make_device(paired=True, connected=True, audio=True, controller=True)
assert device_status_text(device, "connecting", ADDRESS) == "Connecting..."
assert device_status_text(device, "", ADDRESS) == "Connected / audio output / controller"
def test_device_action_policy_matches_the_daemon_onroad_rules():
unpaired = make_device()
paired = make_device(paired=True)
assert device_action_allowed(unpaired, "", offroad=True)
assert not device_action_allowed(unpaired, "", offroad=False)
assert device_action_allowed(paired, "", offroad=False)
assert not device_action_allowed(paired, "pairing", offroad=True)
def test_primary_device_action_is_pair_then_connect_then_manage():
unpaired = make_device()
paired = make_device(paired=True)
connected = make_device(paired=True, connected=True)
manager = FakeBluetoothManager(BluetoothStatus(offroad=True, devices=(unpaired,)))
ui = make_ui(manager)
ui._select_device(ADDRESS)
assert manager.calls == [("pair", ADDRESS)]
manager.status = BluetoothStatus(offroad=True, devices=(paired,))
ui._select_device(ADDRESS)
assert manager.calls[-1] == ("connect", ADDRESS)
managed = []
ui._show_device_actions = lambda device: managed.append(device.address)
manager.status = BluetoothStatus(offroad=True, devices=(connected,))
ui._select_device(ADDRESS)
assert managed == [ADDRESS]
def test_scan_is_only_requested_when_the_existing_daemon_policy_allows_it():
manager = FakeBluetoothManager(BluetoothStatus(enabled=True, offroad=True))
ui = make_ui(manager)
ui._scan()
assert manager.calls == [("scan", True)]
ui._scan()
assert manager.calls == [("scan", True)]
manager.status = BluetoothStatus(enabled=True, offroad=False)
ui._scan()
assert manager.calls == [("scan", True)]
def test_pairing_dialog_callbacks_send_explicit_acceptance_or_rejection():
manager = FakeBluetoothManager(BluetoothStatus())
ui = make_ui(manager)
ui._on_pairing_confirmation("confirm", DialogResult.CANCEL)
ui._on_pairing_confirmation("confirm", DialogResult.CONFIRM)
ui._keyboard = SimpleNamespace(text="123456", clear=lambda: None)
ui._on_pairing_value("pin", DialogResult.CANCEL)
ui._on_pairing_value("pin", DialogResult.CONFIRM)
assert manager.calls == [
("respond", "confirm", False, ""),
("respond", "confirm", True, ""),
("respond", "pin", False, ""),
("respond", "pin", True, "123456"),
]
+122 -13
View File
@@ -1,9 +1,10 @@
import json
import os
import socket
import threading
import time
from dataclasses import asdict, dataclass
from dataclasses import asdict, dataclass, replace
from pathlib import Path
from typing import Any
@@ -24,6 +25,7 @@ COMMAND_TIMEOUTS = {
"forget": 20.0,
"test_audio": 10.0,
}
TRUE_VALUES = {"1", "true", "yes", "on"}
@dataclass(frozen=True)
@@ -99,12 +101,123 @@ def show_pairing_device(address: str, name: str, paired: bool, trusted: bool, co
return known or (named and not blocked and (audio or controller))
class _DesktopFakeBluetooth:
"""Small stateful fallback for desktop UI demos when bluetooth_managerd is absent."""
def __init__(self):
self._lock = threading.Lock()
self._enabled = True
self._discovering = False
self._selected_audio = "00:11:22:33:44:55"
self._devices = (
BluetoothDevice("00:11:22:33:44:55", "Bluetooth Speaker", paired=True, trusted=True, connected=True, audio=True, rssi=-34),
BluetoothDevice("AA:BB:CC:DD:EE:FF", "Game Controller", controller=True, rssi=-48),
BluetoothDevice("10:20:30:40:50:60", "Wireless Headphones", audio=True, rssi=-63),
)
def status(self) -> BluetoothStatus:
with self._lock:
return BluetoothStatus(
available=True,
enabled=self._enabled,
powered=self._enabled,
discovering=self._discovering,
# Desktop demos do not run bluetooth_managerd, so keep the mock usable from Settings.
offroad=True,
selected_audio=self._selected_audio,
devices=self._devices,
)
def _device_index(self, address: str) -> int:
normalized_address = address.upper()
for index, device in enumerate(self._devices):
if device.address.upper() == normalized_address:
return index
raise RuntimeError("Bluetooth device not found")
def _replace_device(self, address: str, **changes: Any) -> BluetoothDevice:
index = self._device_index(address)
device = replace(self._devices[index], **changes)
self._devices = (*self._devices[:index], device, *self._devices[index + 1:])
return device
def _require_enabled(self) -> None:
if not self._enabled:
raise RuntimeError("Enable Bluetooth before continuing")
def call(self, command: str, **payload: Any) -> dict[str, Any]:
with self._lock:
address = str(payload.get("address", ""))
if command == "set_power":
self._enabled = bool(payload.get("enabled", False))
self._discovering = False
if not self._enabled:
self._selected_audio = ""
self._devices = tuple(replace(device, connected=False) for device in self._devices)
elif command == "start_scan":
self._require_enabled()
self._discovering = True
elif command == "stop_scan":
self._discovering = False
elif command == "pair":
self._require_enabled()
self._replace_device(address, paired=True, trusted=True)
elif command == "connect":
self._require_enabled()
device = self._devices[self._device_index(address)]
if not device.paired:
raise RuntimeError("Pair the Bluetooth device before connecting")
self._replace_device(address, connected=True)
elif command == "disconnect":
self._replace_device(address, connected=False)
elif command == "forget":
self._replace_device(address, paired=False, trusted=False, connected=False)
if self._selected_audio.upper() == address.upper():
self._selected_audio = ""
elif command == "select_audio":
if not address:
self._selected_audio = ""
else:
device = self._devices[self._device_index(address)]
if not device.audio:
raise RuntimeError("Selected device does not support Bluetooth audio")
self._selected_audio = device.address
elif command == "test_audio":
device = self._devices[self._device_index(address)]
if not device.audio:
raise RuntimeError("Selected device does not support Bluetooth audio")
if not device.paired or not device.connected:
raise RuntimeError("Connect the Bluetooth audio device before testing")
self._selected_audio = device.address
return {"audio_test_delay_ms": 3000}
elif command != "pairing_response":
raise RuntimeError(f"Unknown Bluetooth command: {command}")
return {}
class BluetoothClient:
def __init__(self, socket_path: str = BLUETOOTH_SOCKET_PATH, timeout: float = 5.0):
self.socket_path = socket_path
self.timeout = timeout
self._desktop_fake: _DesktopFakeBluetooth | None = None
def _get_desktop_fake(self) -> _DesktopFakeBluetooth | None:
if not (os.getenv("SP_ALLOW_DESKTOP_FAKE_BLUETOOTH", "0").lower() in TRUE_VALUES and
os.getenv("SIMULATION", "0").lower() in TRUE_VALUES and
os.getenv("NOBOARD", "0").lower() in TRUE_VALUES and
not os.path.exists(self.socket_path)):
return None
from openpilot.system.hardware import PC
if not PC:
return None
if self._desktop_fake is None:
self._desktop_fake = _DesktopFakeBluetooth()
return self._desktop_fake
def call(self, command: str, **payload: Any) -> dict[str, Any]:
fake = self._get_desktop_fake()
if fake is not None:
return fake.call(command, **payload)
request = json.dumps({"command": command, **payload}, separators=(",", ":")).encode() + b"\n"
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
sock.settimeout(max(self.timeout, COMMAND_TIMEOUTS.get(command, 0.0)))
@@ -125,18 +238,9 @@ class BluetoothClient:
return result
def status(self) -> BluetoothStatus:
if os.getenv("SP_ALLOW_DESKTOP_FAKE_BLUETOOTH", "0") == "1" and not os.path.exists(self.socket_path):
return BluetoothStatus(
available=True,
enabled=True,
powered=True,
discovering=False,
offroad=True,
devices=(
BluetoothDevice("00:11:22:33:44:55", "Bluetooth Speaker", paired=True, connected=True, audio=True),
BluetoothDevice("AA:BB:CC:DD:EE:FF", "Game Controller", controller=True, rssi=-48),
),
)
fake = self._get_desktop_fake()
if fake is not None:
return fake.status()
if not os.path.exists(self.socket_path):
params = Params()
return BluetoothStatus(
@@ -152,6 +256,11 @@ class BluetoothClient:
return asdict(status)
def set_power(self, enabled: bool) -> None:
fake = self._get_desktop_fake()
if fake is not None:
fake.call("set_power", enabled=enabled)
return
params = Params()
bootstrap = enabled and not os.path.exists(self.socket_path)
if bootstrap:
@@ -8,7 +8,9 @@ import pytest
from openpilot.starpilot.system.bluetooth.audio import BluetoothAudioSink
from openpilot.starpilot.system.bluetooth.bluez import PairingAgent
from openpilot.starpilot.system.bluetooth.daemon import BluetoothController
from openpilot.starpilot.system.bluetooth.protocol import A2DP_SINK_UUID, HID_UUID, BluetoothDevice, BluetoothStatus, device_capabilities, show_pairing_device
from openpilot.starpilot.system.bluetooth.protocol import (A2DP_SINK_UUID, HID_UUID, BluetoothClient, BluetoothDevice, BluetoothStatus,
device_capabilities, show_pairing_device)
from openpilot.system import hardware
class FakeParams:
@@ -140,6 +142,52 @@ def test_pairing_list_filters_anonymous_and_irrelevant_advertisements():
assert show_pairing_device("00:11:22:33:44:55", "Known device", True, True, False, False, False, False)
def test_desktop_fake_bluetooth_is_stateful_and_interactive(monkeypatch, tmp_path):
monkeypatch.setenv("SP_ALLOW_DESKTOP_FAKE_BLUETOOTH", "1")
monkeypatch.setenv("SIMULATION", "1")
monkeypatch.setenv("NOBOARD", "1")
client = BluetoothClient(socket_path=str(tmp_path / "bluetooth.sock"))
initial = client.status()
speaker, controller = initial.devices[:2]
assert initial.available and initial.enabled and speaker.connected
client.start_scan()
assert client.status().discovering
client.pair(controller.address)
client.connect(controller.address)
paired_controller = next(device for device in client.status().devices if device.address == controller.address)
assert paired_controller.paired and paired_controller.trusted and paired_controller.connected
client.select_audio(speaker.address)
assert client.status().selected_audio == speaker.address
assert client.test_audio(speaker.address) == 3.0
client.forget(controller.address)
forgotten_controller = next(device for device in client.status().devices if device.address == controller.address)
assert not forgotten_controller.paired and not forgotten_controller.connected
client.set_power(False)
disabled = client.status()
assert not disabled.enabled and not disabled.powered and not disabled.discovering
with pytest.raises(RuntimeError, match="Enable Bluetooth"):
client.start_scan()
client.set_power(True)
assert client.status().enabled
def test_desktop_fake_bluetooth_cannot_activate_on_device(monkeypatch, tmp_path):
monkeypatch.setenv("SP_ALLOW_DESKTOP_FAKE_BLUETOOTH", "1")
monkeypatch.setenv("SIMULATION", "1")
monkeypatch.setenv("NOBOARD", "1")
monkeypatch.setattr(hardware, "PC", False)
client = BluetoothClient(socket_path=str(tmp_path / "bluetooth.sock"))
assert client._get_desktop_fake() is None
assert client._desktop_fake is None
def test_pairing_agent_accept_reject_and_timeout():
agent = PairingAgent()
result = []
+428
View File
@@ -0,0 +1,428 @@
from __future__ import annotations
from dataclasses import dataclass
from functools import partial
import pyray as rl
from openpilot.starpilot.system.bluetooth.protocol import BluetoothDevice, BluetoothStatus
from openpilot.system.ui.lib.application import FontWeight, MousePos, gui_app
from openpilot.system.ui.lib.bluetooth_manager import BluetoothManager
from openpilot.system.ui.lib.multilang import tr
from openpilot.system.ui.lib.scroll_panel import GuiScrollPanel
from openpilot.system.ui.widgets import DialogResult, Widget
from openpilot.system.ui.widgets.button import Button, ButtonStyle
from openpilot.system.ui.widgets.confirm_dialog import ConfirmDialog, alert_dialog
from openpilot.system.ui.widgets.keyboard import Keyboard
from openpilot.system.ui.widgets.label import gui_label
from openpilot.system.ui.widgets.option_dialog import MultiOptionDialog
from openpilot.system.ui.widgets.toggle import Toggle
HEADER_HEIGHT = 180
ITEM_HEIGHT = 160
HEADER_PADDING = 40
SCAN_BUTTON_WIDTH = 260
FORGET_BUTTON_WIDTH = 180
ACTION_GAP = 35
# Match the Network panel: the Settings surface stays pure black and list rows are transparent over it.
PANEL_BACKGROUND = rl.BLACK
DIALOG_BACKGROUND = rl.Color(27, 27, 27, 255)
ROW_BORDER = rl.LIGHTGRAY
TEXT_SECONDARY = rl.Color(170, 170, 170, 255)
TEXT_DISABLED = rl.Color(150, 150, 150, 255)
TEXT_CONNECTED = rl.Color(113, 209, 135, 255)
def device_status_text(device: BluetoothDevice, operation: str, selected_audio: str) -> str:
"""Return the concise, state-first label shown below a Bluetooth device name."""
if operation:
return operation.capitalize() + "..."
capabilities = []
if device.audio:
capabilities.append(tr("audio output") if selected_audio.upper() == device.address.upper() else tr("audio"))
if device.controller:
capabilities.append(tr("controller"))
capability_text = " / ".join(capabilities)
if device.connected:
return tr("Connected") + (f" / {capability_text}" if capability_text else "")
if device.paired:
return tr("Paired - tap to connect")
return tr("Tap to pair") + (f" / {capability_text}" if capability_text else "")
def device_action_allowed(device: BluetoothDevice, operation: str, offroad: bool) -> bool:
"""Mirror the daemon's operation policy before a row can receive a tap."""
if operation:
return False
if not offroad and not device.paired:
return False
return True
@dataclass(frozen=True)
class _DeviceViewState:
device: BluetoothDevice
operation: str
selected_audio: str
offroad: bool
class BluetoothDeviceRow(Widget):
"""A Wi-Fi-style, scroll-safe row with a primary device action and optional forget button."""
def __init__(self, address: str, on_select, on_forget):
super().__init__()
self.address = address
self._on_select = on_select
self._on_forget = on_forget
self._state: _DeviceViewState | None = None
self._forget_button = Button(tr("Forget"), on_forget, button_style=ButtonStyle.FORGET_WIFI, font_size=42)
self._forget_rect = rl.Rectangle(0, 0, 0, 0)
def update(self, state: _DeviceViewState) -> None:
self._state = state
self.set_enabled(device_action_allowed(state.device, state.operation, state.offroad))
def set_touch_valid_callback(self, touch_callback):
super().set_touch_valid_callback(touch_callback)
self._forget_button.set_touch_valid_callback(touch_callback)
def _show_forget(self) -> bool:
state = self._state
return bool(state and state.device.paired and state.offroad and not state.operation)
def _update_layout_rects(self) -> None:
if self._show_forget():
self._forget_rect = rl.Rectangle(
self._rect.x + self._rect.width - FORGET_BUTTON_WIDTH,
self._rect.y + (self._rect.height - 80) / 2,
FORGET_BUTTON_WIDTH,
80,
)
else:
self._forget_rect = rl.Rectangle(0, 0, 0, 0)
def _handle_mouse_release(self, mouse_pos: MousePos) -> None:
if self._show_forget() and rl.check_collision_point_rec(mouse_pos, self._forget_rect):
return
self._on_select()
def _render(self, rect: rl.Rectangle):
state = self._state
if state is None:
return
self._update_layout_rects()
enabled = self.enabled
rl.draw_rectangle_rec(rect, PANEL_BACKGROUND)
right_padding = FORGET_BUTTON_WIDTH + ACTION_GAP if self._show_forget() else HEADER_PADDING
text_rect = rl.Rectangle(rect.x + HEADER_PADDING, rect.y + 18, rect.width - HEADER_PADDING - right_padding, 62)
text_color = rl.WHITE if enabled else TEXT_DISABLED
gui_label(text_rect, state.device.name, font_size=54, color=text_color, font_weight=FontWeight.MEDIUM)
status_rect = rl.Rectangle(text_rect.x, rect.y + 82, text_rect.width, 52)
status = device_status_text(state.device, state.operation, state.selected_audio)
status_color = TEXT_CONNECTED if state.device.connected and not state.operation else TEXT_SECONDARY
if not enabled:
status_color = TEXT_DISABLED
gui_label(status_rect, status, font_size=39, color=status_color)
if self._show_forget():
self._forget_button.set_enabled(enabled)
self._forget_button.set_parent_rect(self._parent_rect or rect)
self._forget_button.render(self._forget_rect)
class BluetoothAudioTestDialog(Widget):
"""Shows the existing delayed Bluetooth audio test without introducing another backend state."""
def __init__(self, manager: BluetoothManager):
super().__init__()
self._manager = manager
self._error = ""
self._done_button = Button(tr("Done"), gui_app.pop_widget, button_style=ButtonStyle.PRIMARY)
def _update_state(self):
if not self._error:
self._error = self._manager.consume_error()
def _render(self, rect: rl.Rectangle):
dialog_rect = rl.Rectangle(rect.x + rect.width * 0.25, rect.y + rect.height * 0.3, rect.width * 0.5, rect.height * 0.4)
rl.draw_rectangle_rounded(dialog_rect, 0.04, 20, DIALOG_BACKGROUND)
if self._error:
title = tr("Bluetooth audio test failed")
detail = self._error
else:
phase = self._manager.audio_test_phase()
title = tr("Bluetooth audio test")
if phase == "NOW":
detail = tr("Playing now")
elif phase == "complete":
detail = tr("Complete")
else:
detail = tr("Playing in {}...").format(phase)
gui_label(rl.Rectangle(dialog_rect.x + 50, dialog_rect.y + 55, dialog_rect.width - 100, 80), title,
font_size=62, font_weight=FontWeight.BOLD, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
gui_label(rl.Rectangle(dialog_rect.x + 50, dialog_rect.y + 155, dialog_rect.width - 100, 80), detail,
font_size=50, color=TEXT_SECONDARY, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
button_rect = rl.Rectangle(dialog_rect.x + 50, dialog_rect.y + dialog_rect.height - 150, dialog_rect.width - 100, 100)
self._done_button.render(button_rect)
class BluetoothManagerUI(Widget):
"""Big UI Bluetooth settings panel backed by the existing Bluetooth manager daemon."""
def __init__(self, manager: BluetoothManager):
super().__init__()
self._manager = manager
self._scroll_panel = GuiScrollPanel()
self._power_toggle = Toggle(initial_state=False, callback=self._toggle_power)
self._scan_button = Button(tr("Scan"), self._scan, button_style=ButtonStyle.NORMAL, font_size=42)
self._device_rows: dict[str, BluetoothDeviceRow] = {}
self._pending_power: bool | None = None
self._scan_pending = False
self._scan_on_ready = False
self._last_prompt_id = ""
self._last_status_error = ""
self._last_operation_error = ""
self._keyboard = Keyboard(max_text_size=32, min_text_size=1, password_mode=False)
def show_event(self):
super().show_event()
self._manager.set_active(True)
self._scan_on_ready = True
def hide_event(self):
status = self._manager.status
if status.discovering and status.offroad:
self._manager.set_scanning(False)
self._scan_pending = False
self._manager.set_active(False)
super().hide_event()
def _toggle_power(self, enabled: bool):
status = self._manager.status
if not status.available or not status.offroad:
self._power_toggle.set_state(status.enabled)
return
self._pending_power = enabled
self._scan_on_ready = enabled
self._manager.set_power(enabled)
def _scan(self):
status = self._manager.status
if status.enabled and status.offroad and not status.discovering and not self._scan_pending:
self._scan_pending = True
self._manager.set_scanning(True)
def _operation_for(self, device: BluetoothDevice, status: BluetoothStatus) -> str:
operation = self._manager.operation_for(device.address)
if not operation and status.pairing_address.upper() == device.address.upper():
operation = tr("pairing")
return operation
def _sync_power(self, status: BluetoothStatus) -> None:
if self._pending_power is not None and status.enabled == self._pending_power:
self._pending_power = None
if status.discovering or not status.enabled:
self._scan_pending = False
if self._pending_power is None:
self._power_toggle.set_state(status.enabled)
self._power_toggle.set_enabled(status.available and status.offroad and self._pending_power is None)
self._scan_button.set_enabled(status.enabled and status.offroad and not status.discovering and not self._scan_pending)
self._scan_button.set_text(tr("Scanning") if status.discovering or self._scan_pending else tr("Scan"))
def _sync_rows(self, status: BluetoothStatus) -> list[BluetoothDeviceRow]:
visible_rows = []
known_addresses = set()
for device in status.devices:
address = device.address.upper()
known_addresses.add(address)
row = self._device_rows.get(address)
if row is None:
row = BluetoothDeviceRow(
address,
on_select=partial(self._select_device, address),
on_forget=partial(self._confirm_forget, address),
)
row.set_touch_valid_callback(self._scroll_panel.is_touch_valid)
self._device_rows[address] = row
row.update(_DeviceViewState(device, self._operation_for(device, status), status.selected_audio, status.offroad))
visible_rows.append(row)
self._device_rows = {address: row for address, row in self._device_rows.items() if address in known_addresses}
return visible_rows
def _select_device(self, address: str):
device = self._find_device(address)
status = self._manager.status
if device is None or not device_action_allowed(device, self._operation_for(device, status), status.offroad):
return
if not device.paired:
self._manager.pair(device.address)
elif not device.connected:
self._manager.connect(device.address)
else:
self._show_device_actions(device)
def _find_device(self, address: str) -> BluetoothDevice | None:
return next((device for device in self._manager.status.devices if device.address.upper() == address.upper()), None)
def _show_device_actions(self, device: BluetoothDevice):
status = self._manager.status
options = [tr("Disconnect")]
if device.audio and status.offroad:
selected = status.selected_audio.upper() == device.address.upper()
options.append(tr("Stop using for audio") if selected else tr("Use for audio"))
options.append(tr("Test audio"))
def apply(result: DialogResult):
if result != DialogResult.CONFIRM:
return
if dialog.selection == tr("Disconnect"):
self._manager.disconnect(device.address)
elif dialog.selection == tr("Use for audio"):
self._manager.select_audio(device.address)
elif dialog.selection == tr("Stop using for audio"):
self._manager.select_audio("")
elif dialog.selection == tr("Test audio"):
self._manager.test_audio(device.address)
gui_app.push_widget(BluetoothAudioTestDialog(self._manager))
dialog = MultiOptionDialog(device.name, options, options[0], callback=apply)
gui_app.push_widget(dialog)
def _confirm_forget(self, address: str):
device = self._find_device(address)
status = self._manager.status
if device is None or not device.paired or not status.offroad:
return
def apply(result: DialogResult):
if result == DialogResult.CONFIRM:
self._manager.forget(device.address)
gui_app.push_widget(ConfirmDialog(tr("Forget Bluetooth device \"{}\"?").format(device.name), tr("Forget"), callback=apply))
def _handle_prompt(self, status: BluetoothStatus):
prompt = status.prompt
if prompt is None:
self._last_prompt_id = ""
return
prompt_id = str(prompt.get("id", ""))
if not prompt_id or prompt_id == self._last_prompt_id:
return
self._last_prompt_id = prompt_id
name = str(prompt.get("name") or tr("Bluetooth device"))
kind = str(prompt.get("kind") or "")
value = str(prompt.get("value") or "")
if prompt.get("display_only"):
message = value if value else tr("Pairing with {}.").format(name)
gui_app.push_widget(alert_dialog(message))
elif kind in ("pin", "passkey"):
self._keyboard.reset(min_text_size=1)
self._keyboard.set_title(tr("Enter {} for {}").format(kind, name), "")
self._keyboard.set_callback(partial(self._on_pairing_value, prompt_id))
gui_app.push_widget(self._keyboard)
else:
message = tr("Pair with {}?").format(name)
if value:
message += f"\n{value}"
gui_app.push_widget(ConfirmDialog(message, tr("Pair"), callback=partial(self._on_pairing_confirmation, prompt_id)))
def _on_pairing_value(self, prompt_id: str, result: DialogResult):
value = self._keyboard.text
self._keyboard.clear()
self._manager.respond(prompt_id, result == DialogResult.CONFIRM, value if result == DialogResult.CONFIRM else "")
def _on_pairing_confirmation(self, prompt_id: str, result: DialogResult):
self._manager.respond(prompt_id, result == DialogResult.CONFIRM)
def _show_errors(self, status: BluetoothStatus) -> bool:
operation_error = self._manager.consume_error()
if operation_error:
self._pending_power = None
self._scan_pending = False
if operation_error != self._last_operation_error:
self._last_operation_error = operation_error
gui_app.push_widget(alert_dialog(operation_error))
return True
return False
self._last_operation_error = ""
if not status.error:
self._last_status_error = ""
return False
if status.error != self._last_status_error:
self._last_status_error = status.error
gui_app.push_widget(alert_dialog(status.error))
return True
return False
def _update_state(self):
status = self._manager.status
self._sync_power(status)
if self._scan_on_ready and status.available and status.enabled:
self._scan_on_ready = False
if status.offroad and not status.discovering:
self._scan()
if not self._show_errors(status):
self._handle_prompt(status)
def _render(self, rect: rl.Rectangle):
status = self._manager.status
header_rect = rl.Rectangle(rect.x, rect.y, rect.width, HEADER_HEIGHT)
self._render_header(header_rect, status)
content_rect = rl.Rectangle(rect.x, rect.y + HEADER_HEIGHT, rect.width, rect.height - HEADER_HEIGHT)
if not status.available:
gui_label(content_rect, tr("Bluetooth is not available on this device."), font_size=62,
color=TEXT_SECONDARY, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
return
if not status.enabled:
gui_label(content_rect, tr("Bluetooth is off."), font_size=62, color=TEXT_SECONDARY,
alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
return
rows = self._sync_rows(status)
if not rows:
message = tr("Scanning for Bluetooth devices...") if status.discovering else tr("No Bluetooth devices found.")
gui_label(content_rect, message, font_size=62, color=TEXT_SECONDARY, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
return
self._render_device_list(content_rect, rows)
def _render_header(self, rect: rl.Rectangle, status: BluetoothStatus):
rl.draw_rectangle_rec(rect, PANEL_BACKGROUND)
line_y = int(rect.y + rect.height - 1)
rl.draw_line(int(rect.x), line_y, int(rect.x + rect.width), line_y, ROW_BORDER)
gui_label(rl.Rectangle(rect.x + HEADER_PADDING, rect.y + 26, 500, 68), tr("Bluetooth"), font_size=64, font_weight=FontWeight.BOLD)
subtitle = tr("On") if status.enabled else tr("Off")
if status.enabled and not status.offroad:
subtitle += " - " + tr("Limited while driving")
gui_label(rl.Rectangle(rect.x + HEADER_PADDING, rect.y + 104, 650, 44), subtitle, font_size=40, color=TEXT_SECONDARY)
toggle_rect = rl.Rectangle(rect.x + rect.width - HEADER_PADDING - 160, rect.y + (rect.height - 80) / 2, 160, 80)
self._power_toggle.render(toggle_rect)
scan_rect = rl.Rectangle(toggle_rect.x - ACTION_GAP - SCAN_BUTTON_WIDTH, rect.y + (rect.height - 100) / 2, SCAN_BUTTON_WIDTH, 100)
self._scan_button.render(scan_rect)
def _render_device_list(self, rect: rl.Rectangle, rows: list[BluetoothDeviceRow]):
content_rect = rl.Rectangle(rect.x, rect.y, rect.width, len(rows) * ITEM_HEIGHT)
offset = self._scroll_panel.update(rect, content_rect)
rl.begin_scissor_mode(int(rect.x), int(rect.y), int(rect.width), int(rect.height))
for index, row in enumerate(rows):
item_rect = rl.Rectangle(rect.x, rect.y + index * ITEM_HEIGHT + offset, rect.width, ITEM_HEIGHT)
if not rl.check_collision_recs(item_rect, rect):
continue
row.set_parent_rect(rect)
row.render(item_rect)
if index < len(rows) - 1:
line_y = int(item_rect.y + item_rect.height - 1)
rl.draw_line(int(item_rect.x), line_y, int(item_rect.x + item_rect.width), line_y, ROW_BORDER)
rl.end_scissor_mode()