petere
eb08cd4282
Current state of main at 0240060 feat(hp-laptop): install TeleportFling from its flake. History intentionally collapsed to a single commit; this repo mirrors only the latest state.
213 lines
6.9 KiB
Python
213 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
|
"""QuickShell bluetooth manager backend.
|
|
|
|
Talks to bluez via `bluetoothctl` (text/shell parsing — the tool has no JSON
|
|
mode). State is read live from the adapter (`bluetoothctl show`) and per device
|
|
(`bluetoothctl info <mac>`), so it reflects reality even after pairing from
|
|
another app. The `qs-bt` wrapper (home-manager hyprland module) passes the
|
|
absolute bluetoothctl path in $BLUECTL.
|
|
|
|
Commands
|
|
--------
|
|
status -> JSON adapter + devices
|
|
{powered,name,alias,pairable,discoverable,
|
|
discovering,devices:[{mac,name,icon,paired,
|
|
trusted,connected,battery}]}
|
|
power on|off -> toggle adapter power -> JSON ok
|
|
scan -> discovery for SCAN_SECONDS, then stop; JSON
|
|
list of {mac,name} found during the scan
|
|
connect <mac> / disconnect <mac>
|
|
pair <mac> -> pair (bounded; pairing prompts end it)
|
|
trust <mac> / untrust <mac>
|
|
remove <mac> -> unpair / forget
|
|
All non-status commands return {"ok": bool, "error": "..."}.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
|
|
BLUECTL = os.environ.get("BLUECTL", "bluetoothctl")
|
|
SCAN_SECONDS = 8
|
|
PAIR_TIMEOUT = 30
|
|
CONNECT_TIMEOUT = 12
|
|
WAIT_TIMEOUT = 10
|
|
|
|
|
|
def run(args, timeout=WAIT_TIMEOUT):
|
|
try:
|
|
p = subprocess.run(
|
|
[BLUECTL] + args,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
)
|
|
return p.returncode, p.stdout, p.stderr
|
|
except subprocess.TimeoutExpired as exc:
|
|
tail = (exc.stdout or exc.stderr or b"")
|
|
if isinstance(tail, bytes):
|
|
tail = tail.decode(errors="replace")
|
|
return -1, tail, "command timed out"
|
|
except OSError as exc:
|
|
return -2, "", str(exc)
|
|
|
|
|
|
def field(line, key):
|
|
low = line.lstrip()
|
|
return low.split(":", 1)[1].strip() if low.startswith(key + ":") else None
|
|
|
|
|
|
def parse_adapter(out):
|
|
"""Adapter sections from `bluetoothctl show`; prefer the [default] one."""
|
|
sections = []
|
|
cur = None
|
|
for line in out.splitlines():
|
|
if line.startswith("Controller "):
|
|
bits = line.split()
|
|
cur = {"mac": bits[1], "default": "[default]" in line}
|
|
sections.append(cur)
|
|
elif cur is not None:
|
|
v = field(line, "Powered")
|
|
if v:
|
|
cur["powered"] = v == "yes"
|
|
v = field(line, "Discovering")
|
|
if v:
|
|
cur["discovering"] = v == "yes"
|
|
v = field(line, "Pairable")
|
|
if v:
|
|
cur["pairable"] = v == "yes"
|
|
v = field(line, "Discoverable")
|
|
if v:
|
|
cur["discoverable"] = v == "yes"
|
|
v = field(line, "Name")
|
|
if v:
|
|
cur["name"] = v
|
|
v = field(line, "Alias")
|
|
if v:
|
|
cur["alias"] = v
|
|
for s in sections:
|
|
if s.get("default"):
|
|
return s
|
|
return sections[0] if sections else None
|
|
|
|
|
|
def parse_device(mac):
|
|
rc, out, _ = run(["info", mac], timeout=WAIT_TIMEOUT)
|
|
dev = {
|
|
"mac": mac,
|
|
"name": "",
|
|
"icon": "",
|
|
"paired": False,
|
|
"trusted": False,
|
|
"connected": False,
|
|
"battery": None,
|
|
}
|
|
if rc != 0:
|
|
return dev
|
|
for line in out.splitlines():
|
|
v = field(line, "Name")
|
|
if v:
|
|
dev["name"] = v
|
|
v = field(line, "Icon")
|
|
if v:
|
|
dev["icon"] = v
|
|
v = field(line, "Paired")
|
|
if v:
|
|
dev["paired"] = v == "yes"
|
|
v = field(line, "Trusted")
|
|
if v:
|
|
dev["trusted"] = v == "yes"
|
|
v = field(line, "Connected")
|
|
if v:
|
|
dev["connected"] = v == "yes"
|
|
m = re.search(r"Battery Percentage:\s*0x[0-9A-Fa-f]+\s*\((\d+)\)", line)
|
|
if m:
|
|
dev["battery"] = int(m.group(1))
|
|
return dev
|
|
|
|
|
|
def list_devices():
|
|
rc, out, _ = run(["devices"], timeout=WAIT_TIMEOUT)
|
|
macs = []
|
|
if rc == 0:
|
|
for line in out.splitlines():
|
|
bits = line.split()
|
|
if len(bits) >= 2 and bits[0] == "Device":
|
|
mac = bits[1]
|
|
if mac not in macs:
|
|
macs.append(mac)
|
|
return [parse_device(m) for m in macs]
|
|
|
|
|
|
def status():
|
|
rc, out, _ = run(["show"], timeout=WAIT_TIMEOUT)
|
|
adapter = parse_adapter(out) if rc == 0 else None
|
|
devs = list_devices()
|
|
# connected first, then alphabetical — the stable order the popup renders.
|
|
devs.sort(key=lambda d: (not d["connected"], (d["name"] or d["mac"]).lower()))
|
|
return {
|
|
"powered": bool(adapter and adapter.get("powered")),
|
|
"name": adapter.get("name") if adapter else "",
|
|
"alias": adapter.get("alias") if adapter else "",
|
|
"pairable": bool(adapter and adapter.get("pairable")),
|
|
"discoverable": bool(adapter and adapter.get("discoverable")),
|
|
"discovering": bool(adapter and adapter.get("discovering")),
|
|
"devices": devs,
|
|
}
|
|
|
|
|
|
def scan():
|
|
"""Discovery for SCAN_SECONDS; bluetoothctl parses NEW/CHG device lines."""
|
|
rc, out, _ = run(
|
|
["--timeout", str(SCAN_SECONDS), "scan", "on"],
|
|
timeout=SCAN_SECONDS + 8,
|
|
)
|
|
found = {}
|
|
for line in out.splitlines():
|
|
if line.startswith("[NEW] Device"):
|
|
mac, _, name = line.split()[2:5]
|
|
found[mac] = name
|
|
elif line.startswith("[CHG] Device"):
|
|
bits = line.split()
|
|
if len(bits) >= 5 and bits[3] == "Name:":
|
|
found[bits[2]] = line.split("Name:", 1)[1].strip()
|
|
return [{"mac": m, "name": n} for m, n in sorted(found.items())]
|
|
|
|
|
|
def action(args):
|
|
rc, out, err = run(args, timeout=PAIR_TIMEOUT if args and args[0] == "pair" else CONNECT_TIMEOUT)
|
|
if rc == 0:
|
|
return {"ok": True, "error": ""}
|
|
# bluetoothctl prints failures like "Failed to connect: ..." to stdout.
|
|
detail = ""
|
|
for line in (out or "").splitlines():
|
|
if "Failed" in line or "not" in line.lower() or "error" in line.lower():
|
|
detail = line.strip()
|
|
break
|
|
return {"ok": False, "error": detail or (err or "command failed")}
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
cmd = args[0] if args else "status"
|
|
if cmd == "status":
|
|
print(json.dumps(status()))
|
|
elif cmd == "power":
|
|
val = args[1] if len(args) > 1 else ""
|
|
print(json.dumps(action(["power", val]) if val in ("on", "off") else {"ok": False, "error": "power on|off"}))
|
|
elif cmd == "scan":
|
|
print(json.dumps(scan()))
|
|
elif cmd in ("pair", "connect", "disconnect", "trust", "untrust", "remove"):
|
|
mac = args[1] if len(args) > 1 else ""
|
|
if not mac:
|
|
print(json.dumps({"ok": False, "error": "missing device address"}))
|
|
else:
|
|
print(json.dumps(action([cmd, mac])))
|
|
else:
|
|
print(json.dumps({"ok": False, "error": "unknown command: " + cmd}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |