i3-companion: revert back to using rfcomm to get battery

Unfortunately, Pulseaudio 15.99 is not stable yet.

bin/i3-companion

@@ -739,16 +739,18 @@ async def bluetooth_notifications(i3, event, path, interface, changed, invalid):
             and "Connected" in args[1]
             or args[0] == "org.bluez.Adapter1"
             and "Powered" in args[1]
-            or args[0] == "org.bluez.Battery1"
-            and "Percentage" in args[1]
         ),
     ),
 )
+@static(scheduled=None)
 @retry(2)
-@debounce(4)
+@debounce(0.2)
 @polybar("bluetooth")
 async def bluetooth_status(i3, event, *args):
     """Update bluetooth status for Polybar."""
+    if bluetooth_status.scheduled:
+        bluetooth_status.scheduled.cancel()
+        bluetooth_status.scheduled = None
     if event is StartEvent:
         # Do we have a bluetooth device?
         if not os.path.exists("/sys/class/bluetooth"):
@@ -770,18 +772,17 @@ async def bluetooth_status(i3, event, *args):
                 powered = True
         elif "org.bluez.Device1" in interfaces:
             # We have a device!
-            device = types.SimpleNamespace(major=0, minor=0, battery=None)
+            device = interfaces["org.bluez.Device1"]
-            interface = interfaces["org.bluez.Device1"]
+            if not device["Connected"][1]:
-            if not interface["Connected"][1]:
                 continue
             try:
-                device_class = interface["Class"][1]
+                device_class = device["Class"][1]
-                device.major = (device_class & 0x1F00) >> 8
+                major = (device_class & 0x1F00) >> 8
-                device.minor = (device_class & 0xFC) >> 2
+                minor = (device_class & 0xFC) >> 2
-                device.battery = interfaces["org.bluez.Battery1"]["Percentage"][1]
+                mac = device["Address"][1]
+                devices.append((major, minor, mac))
             except KeyError:
-                pass
+                devices.append((0, 0, ""))
-            devices.append(device)
 
     # Choose appropriate icons for output
     # See: https://btprodspecificationrefs.blob.core.windows.net/assigned-numbers
@@ -790,7 +791,7 @@ async def bluetooth_status(i3, event, *args):
         output = ""
     else:
         output = ["bluetooth"]
-        for device in devices:
+        for major, minor, mac in devices:
             classes = {
                 1: "laptop",
                 2: "phone",
@@ -816,20 +817,54 @@ async def bluetooth_status(i3, event, *args):
                     (lambda x: x & 0x20, "printer"),
                 ],
             }
-            icon = classes.get((device.major, device.minor)) or classes.get(
+            icon = classes.get((major, minor)) or classes.get(major, "unknown")
-                device.major, "unknown"
-            )
             if type(icon) is list:
                 for matcher, name in icon:
-                    if matcher(device.minor):
+                    if matcher(minor):
                         icon = name
                         break
                 else:
                     icon = "unknown"
             output.append(icon)
-            if device.battery is not None:
+            if mac.startswith("2C:41:A1"):
-                icon = f"battery-{(device.battery+12)//25*25}"
+                # Get battery status for BOSE QC 35 II. This is hacky.
+                # For other headsets, the information could be
+                # available through upower DBus. See
+                # https://github.com/Denton-L/based-connect. Starting
+                # from PA 16, it may be exposed by PA, then to BlueZ.
+                # See
+                # https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/merge_requests/482
+                loop = asyncio.get_event_loop()
+                sock = socket.socket(
+                    socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM
+                )
+                try:
+                    sock.setblocking(False)
+                    # Workaround a bug in asyncio: https://bugs.python.org/issue27929
+                    fut = loop.create_future()
+                    loop._sock_connect(fut, sock, (mac, 8))
+                    await fut
+                    # Get battery
+                    for send, ack, size in (
+                        (b"\0\1\1\0", b"\0\1\3\5", 5),  # get firmware
+                        (b"\2\2\1\0", b"\2\2\3\1", 1),  # get battery
+                    ):
+                        await loop.sock_sendall(sock, send)
+                        rack = await loop.sock_recv(sock, len(ack))
+                        assert rack == ack, "incorrect ack received"
+                        result = await loop.sock_recv(sock, size)
+                    battery = result[0]
+                finally:
+                    sock.close()
+                # Choose an icon
+                icon = f"battery-{(battery+12)//25*25}"
                 output[-1] = (output[-1], icon)
+
+                # Schedule a refresh in 5 minutes
+                bluetooth_status.scheduled = loop.call_later(
+                    600,
+                    lambda: asyncio.create_task(bluetooth_status(i3, StartEvent)),
+                )
         return "|".join(
             (" ".join(icons[oo] for oo in o) if type(o) is tuple else icons[o])
             for o in output