99admandClaude Opus 5.5
fc15f2b933
Name, Seitentitel, Dienste (playwish-web, playwish-controller), Dienst-Benutzer (playwish), Pfade (/opt/playwish, /usr/local/src/playwish), Vorlagen und Repository-Adresse. „Audiofy“ war bereits von mehreren Produkten belegt. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
111 lines
4.2 KiB
Python
111 lines
4.2 KiB
Python
"""Playwish – Bluetooth-Lautsprecher suchen, koppeln und verbinden (bluetoothctl, ohne sudo).
|
||
|
||
Gekoppelt wird immer MIT Agent in derselben bluetoothctl-Sitzung – sonst speichert
|
||
BlueZ den Schlüssel nicht dauerhaft ("Bonded: no"), und der Lautsprecher lässt sich
|
||
nach dem Aus- und Einschalten nicht mehr verbinden ("Permission denied").
|
||
"""
|
||
|
||
import logging
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
|
||
log = logging.getLogger("playwish.lautsprecher")
|
||
|
||
MAC_RE = re.compile(r"^([0-9A-F]{2}:){5}[0-9A-F]{2}$")
|
||
SUCHDAUER_S = 15
|
||
MAX_GERAETE = 30
|
||
|
||
|
||
def verfuegbar() -> bool:
|
||
return shutil.which("bluetoothctl") is not None
|
||
|
||
|
||
def _ctl(*args: str, timeout: int = 20) -> str:
|
||
try:
|
||
return subprocess.run(["bluetoothctl", *args], capture_output=True, text=True, timeout=timeout).stdout
|
||
except (OSError, subprocess.SubprocessError) as e:
|
||
log.warning("bluetoothctl %s fehlgeschlagen: %s", args[0] if args else "", e)
|
||
return ""
|
||
|
||
|
||
def info(mac: str) -> dict:
|
||
"""Name und Zustand eines Geräts; "audio" = kann Musik abspielen (A2DP-Sink)."""
|
||
text = _ctl("info", mac, timeout=10)
|
||
name = re.search(r"^\s*Name:\s*(.+)$", text, re.M)
|
||
ja = lambda feld: re.search(rf"^\s*{feld}:\s*yes", text, re.M) is not None # noqa: E731
|
||
return {
|
||
"mac": mac,
|
||
"name": (name.group(1).strip() if name else "")[:100],
|
||
"paired": ja("Paired"),
|
||
"bonded": ja("Bonded"),
|
||
"connected": ja("Connected"),
|
||
"audio": "Audio Sink" in text or re.search(r"^\s*Icon:\s*audio", text, re.M) is not None,
|
||
}
|
||
|
||
|
||
def oeffentlich(geraet: dict) -> dict:
|
||
"""Nur die Felder, die an den Server gemeldet werden."""
|
||
return {k: geraet[k] for k in ("mac", "name", "paired", "connected")}
|
||
|
||
|
||
def suchen(dauer: int = SUCHDAUER_S) -> list[dict]:
|
||
"""Sucht Bluetooth-Geräte und liefert nur Audiogeräte (verbundene und gekoppelte zuerst)."""
|
||
_ctl("--timeout", str(dauer), "scan", "on", timeout=dauer + 10)
|
||
macs = re.findall(r"^Device\s+(([0-9A-F]{2}:){5}[0-9A-F]{2})", _ctl("devices"), re.M)
|
||
geraete = [g for g in (info(m[0]) for m in macs) if g["audio"]]
|
||
geraete.sort(key=lambda g: (not g["connected"], not g["paired"], g["name"].casefold()))
|
||
return geraete[:MAX_GERAETE]
|
||
|
||
|
||
def trennen(mac: str) -> None:
|
||
_ctl("disconnect", mac, timeout=15)
|
||
|
||
|
||
def verbinden(mac: str) -> bool:
|
||
_ctl("--timeout", "15", "connect", mac, timeout=25)
|
||
return info(mac)["connected"]
|
||
|
||
|
||
def koppeln(mac: str) -> tuple[bool, str]:
|
||
"""Koppelt (falls nötig, mit Agent), vertraut und verbindet. Liefert (ok, Meldung)."""
|
||
if not MAC_RE.match(mac):
|
||
return False, "Ungültige Adresse"
|
||
vorher = info(mac)
|
||
if not vorher["bonded"]:
|
||
# Eine interaktive Sitzung, damit der Agent während des Koppelns registriert bleibt
|
||
befehle = []
|
||
if vorher["paired"]:
|
||
befehle.append((f"remove {mac}", 2)) # halbe Kopplung ohne Schlüssel verwerfen
|
||
befehle += [("agent NoInputNoOutput", 1), ("default-agent", 1), ("pairable on", 1),
|
||
("scan on", 8), ("scan off", 1), (f"pair {mac}", 12), (f"trust {mac}", 2),
|
||
(f"connect {mac}", 10), ("quit", 1)]
|
||
p = None
|
||
try:
|
||
p = subprocess.Popen(["bluetoothctl"], stdin=subprocess.PIPE, stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL, text=True)
|
||
for befehl, warten in befehle:
|
||
p.stdin.write(befehl + "\n")
|
||
p.stdin.flush()
|
||
time.sleep(warten)
|
||
p.stdin.close()
|
||
p.wait(timeout=10)
|
||
except (OSError, subprocess.SubprocessError, BrokenPipeError) as e:
|
||
return False, f"Kopplung fehlgeschlagen ({type(e).__name__})"
|
||
finally:
|
||
if p is not None and p.poll() is None:
|
||
p.kill()
|
||
else:
|
||
_ctl("trust", mac)
|
||
|
||
jetzt = info(mac)
|
||
if not jetzt["connected"] and jetzt["bonded"]:
|
||
verbinden(mac)
|
||
jetzt = info(mac)
|
||
if jetzt["connected"] and jetzt["bonded"]:
|
||
return True, f"Verbunden mit {jetzt['name'] or mac}"
|
||
if not jetzt["bonded"]:
|
||
return False, "Kopplung fehlgeschlagen – ist der Lautsprecher im Kopplungsmodus?"
|
||
return False, "Gekoppelt, aber keine Verbindung – Lautsprecher eingeschaltet und in Reichweite?"
|