bin/yambar-bt.py

binyambar-bt.py

#!/usr/bin/python3

import re
import signal
import subprocess
import sys
import time

import pyudev

con_re = re.compile(r".*Connected: (.*)")
dev_re = re.compile(r"Device (..:..:..:..:..:..) (.*)")

# mac -> name
known_devices = {}

def refresh_known_devices():
    p = subprocess.run(["bluetoothctl"], input="devices", 
            capture_output=True, encoding="utf-8")
    out = p.stdout
    for line in out.split("\n"):
        m = dev_re.match(line)
        if m:
            known_devices[m.group(1)] = m.group(2)

def get_connected_devices():
    connected = []

    if not known_devices:
       refresh_known_devices()

    cmd = ""
    for mac in known_devices:
        cmd += f"info {mac}\n"

    p = subprocess.run(["bluetoothctl"], input=cmd, capture_output=True,
            encoding="utf-8")
    out = p.stdout
    mac = None
    for line in out.split("\n"):
        m = dev_re.match(line)
        if m:
            # New device
            mac = m.group(1)
            continue

        m = con_re.match(line)
        if m:
            # Update connected state
            state = m.group(1)
            if state == "yes":
                connected.append(mac)
    return connected

def print_update(macs):
    status = "disconnected"
    name = "unknown"
    if macs:
        status = "connected"
        name = ""
        names = []
        for mac in macs:
            n = known_devices.get(mac)
            if n:
                names.append(n)
        name = ",".join(names)
    print(f"status|string|{status}")
    print(f"name|string|{name}")
    print()
    sys.stdout.flush()

def signal_handler(sig, frame):
    sys.exit(0)

if __name__ == "__main__":
    # Handle SIGINT/SIGTERM gracefully
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    # Populate our list first
    current_devices = get_connected_devices()
    print_update(current_devices)
    
    context = pyudev.Context()
    monitor = pyudev.Monitor.from_netlink(context)
    monitor.filter_by('bluetooth')

    # Start the udev polling
    for device in iter(monitor.poll, None):
        if device.action == 'add':
            # query bluetoothctl with a small delay to let it settle
            time.sleep(1)
            current_devices = get_connected_devices()
            print_update(current_devices)
        elif device.action == 'remove':
            time.sleep(1)
            current_devices = get_connected_devices()
            # Removed last connected device, output first one we find connected
            print_update(current_devices)