293 lines
11 KiB
Python
293 lines
11 KiB
Python
"""Bluetooth quad: powered by AstalBluetooth (bluez wrapper).
|
|
|
|
Discovery, connect, disconnect and a local connection history (bluez keeps none, so
|
|
we record successful connects in ~/.cache/astro-menu/bt-history.json). 'discovery'
|
|
and 'history' are per-quad feature toggles.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "4.0")
|
|
from gi.repository import GLib, Gtk # noqa: E402
|
|
|
|
from lib.proc import run_text
|
|
from module_base import Feature, ModuleContext, ModuleInstance, ModuleSpec
|
|
from paths import CACHE_DIR
|
|
|
|
_HISTORY = CACHE_DIR / "bt-history.json"
|
|
|
|
|
|
def _load_history() -> list[dict]:
|
|
try:
|
|
return json.loads(_HISTORY.read_text())
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
return []
|
|
|
|
|
|
def _record_history(address: str, name: str) -> None:
|
|
hist = [h for h in _load_history() if h.get("address") != address]
|
|
hist.insert(0, {"address": address, "name": name, "ts": int(time.time())})
|
|
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
_HISTORY.write_text(json.dumps(hist[:50], indent=2))
|
|
|
|
|
|
class _BluetoothView(Gtk.Box):
|
|
def __init__(self, ctx: ModuleContext, full: bool):
|
|
super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=8)
|
|
self.add_css_class("bt-view")
|
|
self.ctx = ctx
|
|
self.full = full
|
|
self.bt = ctx.services.bluetooth
|
|
self._recorded: set[str] = set()
|
|
self._hooked: set[str] = set() # devices whose state signals we've wired
|
|
self._pending: set[str] = set() # addresses with an in-flight connect
|
|
self._failed: set[str] = set() # addresses whose last connect failed
|
|
|
|
self.append(self._build_header())
|
|
|
|
self._list = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4)
|
|
self._list.add_css_class("bt-list")
|
|
if full:
|
|
scroller = Gtk.ScrolledWindow(vexpand=True,
|
|
hscrollbar_policy=Gtk.PolicyType.NEVER)
|
|
scroller.set_child(self._list)
|
|
self.append(scroller)
|
|
else:
|
|
self.append(self._list)
|
|
|
|
if self.bt is not None:
|
|
self.bt.connect("notify::devices", lambda *_: self._refresh())
|
|
self.bt.connect("notify::is-powered", lambda *_: self._refresh())
|
|
self._refresh()
|
|
|
|
def _adapter(self):
|
|
return self.bt.get_adapter() if self.bt else None
|
|
|
|
def _build_header(self) -> Gtk.Widget:
|
|
header = Gtk.CenterBox()
|
|
header.add_css_class("bt-header")
|
|
left = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
|
left.append(Gtk.Label(label="Power"))
|
|
self._power = Gtk.Switch(active=bool(self.bt and self.bt.get_is_powered()),
|
|
valign=Gtk.Align.CENTER)
|
|
self._power.connect("state-set", self._on_power)
|
|
left.append(self._power)
|
|
header.set_start_widget(left)
|
|
|
|
if self.full and self.ctx.feature("discovery", True):
|
|
self._scan = Gtk.ToggleButton(label=" Scan")
|
|
self._scan.add_css_class("quad-action")
|
|
self._scan.connect("toggled", self._on_scan)
|
|
header.set_end_widget(self._scan)
|
|
return header
|
|
|
|
def _on_power(self, _sw, value: bool) -> bool:
|
|
ad = self._adapter()
|
|
if value:
|
|
# The adapter is frequently rfkill soft-blocked at boot; while blocked,
|
|
# AstalBluetooth.set_powered(True) is a silent no-op. Unblock first, then
|
|
# power on (bluez usually auto-powers on unblock, so set_powered is a
|
|
# belt-and-suspenders follow-up).
|
|
run_text(["rfkill", "unblock", "bluetooth"],
|
|
lambda *_a: ad.set_powered(True) if ad else None)
|
|
elif ad:
|
|
ad.set_powered(False)
|
|
return False
|
|
|
|
def _on_scan(self, btn: Gtk.ToggleButton) -> None:
|
|
ad = self._adapter()
|
|
if not ad:
|
|
return
|
|
if btn.get_active():
|
|
ad.start_discovery()
|
|
else:
|
|
ad.stop_discovery()
|
|
|
|
# -- device list -------------------------------------------------------
|
|
def _refresh(self) -> None:
|
|
child = self._list.get_first_child()
|
|
while child:
|
|
self._list.remove(child)
|
|
child = self._list.get_first_child()
|
|
if not self.bt:
|
|
self._list.append(Gtk.Label(label="No System Radio hardware"))
|
|
return
|
|
|
|
devices = list(self.bt.get_devices())
|
|
devices.sort(key=lambda d: (not d.get_connected(), not d.get_paired(),
|
|
(d.get_name() or d.get_address() or "").lower()))
|
|
if not self.full:
|
|
devices = [d for d in devices if d.get_connected() or d.get_paired()][:4]
|
|
for dev in devices:
|
|
self._hook_device(dev)
|
|
self._list.append(self._device_row(dev))
|
|
|
|
if self.full and self.ctx.feature("history", True):
|
|
self._list.append(self._history_section())
|
|
|
|
def _hook_device(self, dev) -> None:
|
|
# React to a device's own state changes (adapter-level notify::devices doesn't
|
|
# fire for per-device connect/disconnect). Hook once each.
|
|
key = dev.get_address() or ""
|
|
if key in self._hooked:
|
|
return
|
|
self._hooked.add(key)
|
|
for sig in ("notify::connected", "notify::connecting", "notify::paired"):
|
|
dev.connect(sig, lambda *_a, d=dev: self._on_device_state(d))
|
|
|
|
def _on_device_state(self, dev) -> None:
|
|
addr = dev.get_address() or ""
|
|
if dev.get_connected():
|
|
self._pending.discard(addr)
|
|
self._failed.discard(addr)
|
|
if self.ctx.feature("history", True):
|
|
self._maybe_record(dev, force=True)
|
|
elif addr in self._pending and not dev.get_connecting():
|
|
# bluez stopped trying without establishing a link → the connect failed
|
|
self._pending.discard(addr)
|
|
self._failed.add(addr)
|
|
self._refresh()
|
|
|
|
def _device_row(self, dev) -> Gtk.Widget:
|
|
row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10)
|
|
row.add_css_class("bt-row")
|
|
addr = dev.get_address() or ""
|
|
icon = Gtk.Image.new_from_icon_name((dev.get_icon() or "bluetooth") + "-symbolic")
|
|
row.append(icon)
|
|
name = dev.get_name() or addr or "Unknown"
|
|
|
|
connected = dev.get_connected()
|
|
connecting = dev.get_connecting() or addr in self._pending
|
|
failed = addr in self._failed
|
|
|
|
lbl = Gtk.Label(label=name, xalign=0.0, hexpand=True)
|
|
row.append(lbl)
|
|
|
|
status = ("connected" if connected else "connecting…" if connecting else
|
|
"failed" if failed else "paired" if dev.get_paired() else "")
|
|
if status:
|
|
tag = Gtk.Label(label=status)
|
|
tag.add_css_class("bt-status")
|
|
if failed:
|
|
tag.add_css_class("bt-failed")
|
|
row.append(tag)
|
|
|
|
actions = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
|
|
if connecting:
|
|
spinner = Gtk.Spinner(valign=Gtk.Align.CENTER)
|
|
spinner.start()
|
|
actions.append(spinner)
|
|
actions.append(self._act("Cancel", lambda: self._cancel(dev)))
|
|
else:
|
|
if dev.get_paired():
|
|
actions.append(self._act("Forget", lambda: self._forget(dev)))
|
|
if connected:
|
|
actions.append(self._act("Disconnect",
|
|
lambda: dev.disconnect_device(None, self._noop)))
|
|
else:
|
|
actions.append(self._act("Retry" if failed else "Connect",
|
|
lambda: self._start_connect(dev)))
|
|
row.append(actions)
|
|
|
|
if connected and self.ctx.feature("history", True):
|
|
self._maybe_record(dev)
|
|
return row
|
|
|
|
@staticmethod
|
|
def _act(label: str, cb) -> Gtk.Button:
|
|
b = Gtk.Button(label=label, valign=Gtk.Align.CENTER)
|
|
b.add_css_class("quad-action")
|
|
b.connect("clicked", lambda *_a: cb())
|
|
return b
|
|
|
|
# -- connect / cancel / forget ----------------------------------------
|
|
def _start_connect(self, dev) -> None:
|
|
addr = dev.get_address() or ""
|
|
self._failed.discard(addr)
|
|
self._pending.add(addr)
|
|
|
|
def done(d, res):
|
|
try:
|
|
d.connect_device_finish(res)
|
|
except Exception:
|
|
a = d.get_address() or ""
|
|
if not d.get_connected():
|
|
self._pending.discard(a)
|
|
self._failed.add(a)
|
|
self._refresh()
|
|
dev.connect_device(None, done)
|
|
# Backstop: some failures never resolve the async call, so time out.
|
|
GLib.timeout_add_seconds(25, lambda: self._connect_timeout(addr))
|
|
self._refresh()
|
|
|
|
def _connect_timeout(self, addr: str) -> bool:
|
|
if addr in self._pending:
|
|
self._pending.discard(addr)
|
|
self._failed.add(addr)
|
|
self._refresh()
|
|
return GLib.SOURCE_REMOVE
|
|
|
|
def _cancel(self, dev) -> None:
|
|
addr = dev.get_address() or ""
|
|
self._pending.discard(addr)
|
|
dev.disconnect_device(None, self._noop) # abort the in-flight attempt
|
|
self._refresh()
|
|
|
|
def _forget(self, dev) -> None:
|
|
ad = self._adapter()
|
|
addr = dev.get_address() or ""
|
|
self._pending.discard(addr)
|
|
self._failed.discard(addr)
|
|
self._hooked.discard(addr)
|
|
if ad:
|
|
ad.remove_device(dev) # fires notify::devices → refresh
|
|
self._refresh()
|
|
|
|
def _maybe_record(self, dev, force: bool = False) -> None:
|
|
addr = dev.get_address() or ""
|
|
if not addr or (addr in self._recorded and not force):
|
|
return
|
|
self._recorded.add(addr)
|
|
_record_history(addr, dev.get_name() or addr)
|
|
|
|
@staticmethod
|
|
def _noop(obj, res) -> None:
|
|
try:
|
|
obj.disconnect_device_finish(res)
|
|
except Exception:
|
|
pass
|
|
|
|
def _history_section(self) -> Gtk.Widget:
|
|
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4)
|
|
box.add_css_class("bt-history")
|
|
box.append(Gtk.Separator())
|
|
box.append(Gtk.Label(label="History", xalign=0.0))
|
|
for h in _load_history()[:8]:
|
|
when = time.strftime("%d.%m %H:%M", time.localtime(h.get("ts", 0)))
|
|
box.append(Gtk.Label(label=f"{h.get('name', '?')} · {when}", xalign=0.0))
|
|
return box
|
|
|
|
|
|
def build(ctx: ModuleContext) -> ModuleInstance:
|
|
# Feature toggles (discovery/history) rebuild the whole card via QuadCard, so
|
|
# both views are recreated with the current features — no partial refresh here.
|
|
compact = _BluetoothView(ctx, full=False)
|
|
expanded = _BluetoothView(ctx, full=True)
|
|
return ModuleInstance(compact=compact, expanded=expanded)
|
|
|
|
|
|
SPEC = ModuleSpec(
|
|
id="bluetooth",
|
|
title="System Radio",
|
|
icon="", # nf-fa-bluetooth
|
|
build=build,
|
|
default_enabled=True,
|
|
features=[Feature("discovery", "Device discovery", True),
|
|
Feature("history", "Connection history", True)],
|
|
)
|