diff --git a/scripts/launch_onroad_desktop.sh b/scripts/launch_onroad_desktop.sh index 3e5249781..6cca1b28b 100755 --- a/scripts/launch_onroad_desktop.sh +++ b/scripts/launch_onroad_desktop.sh @@ -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 diff --git a/scripts/launch_ui_c3_desktop.sh b/scripts/launch_ui_c3_desktop.sh index a41229f0c..88ebebd32 100755 --- a/scripts/launch_ui_c3_desktop.sh +++ b/scripts/launch_ui_c3_desktop.sh @@ -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" diff --git a/selfdrive/ui/layouts/settings/settings.py b/selfdrive/ui/layouts/settings/settings.py index 695b6f193..22771d9fa 100644 --- a/selfdrive/ui/layouts/settings/settings.py +++ b/selfdrive/ui/layouts/settings/settings.py @@ -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()), diff --git a/selfdrive/ui/tests/test_bluetooth_big_ui.py b/selfdrive/ui/tests/test_bluetooth_big_ui.py new file mode 100644 index 000000000..ce2fbea05 --- /dev/null +++ b/selfdrive/ui/tests/test_bluetooth_big_ui.py @@ -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"), + ] diff --git a/starpilot/system/bluetooth/protocol.py b/starpilot/system/bluetooth/protocol.py index ac23bac87..e2641691c 100644 --- a/starpilot/system/bluetooth/protocol.py +++ b/starpilot/system/bluetooth/protocol.py @@ -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: diff --git a/starpilot/system/bluetooth/tests/test_bluetooth.py b/starpilot/system/bluetooth/tests/test_bluetooth.py index 01f550145..03ef72ebf 100644 --- a/starpilot/system/bluetooth/tests/test_bluetooth.py +++ b/starpilot/system/bluetooth/tests/test_bluetooth.py @@ -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 = [] diff --git a/system/ui/widgets/bluetooth.py b/system/ui/widgets/bluetooth.py new file mode 100644 index 000000000..def4d6182 --- /dev/null +++ b/system/ui/widgets/bluetooth.py @@ -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()