fix(mpris): guard GLib.Variant type check before get_type_string

feat(bluetooth): add rfkill soft-block check and BluetoothState enum
This commit is contained in:
sankalp committed 2026-07-21 10:24:39 +05:45
1 parent 301ed9a717
commit 8ac26d6c26
2 files changed
+44 -2

No files matched your search

+43 -1
View File
@@ -1,3 +1,5 @@
import enum
import subprocess
from collections.abc import Callable
from typing import Any, Concatenate, ParamSpec
@@ -13,6 +15,14 @@ BLUEZ_BATTERY_IFACE = "org.bluez.Battery1"
DBUS_OM_IFACE = "org.freedesktop.DBus.ObjectManager"
DBUS_PROPS_IFACE = "org.freedesktop.DBus.Properties"
RFKILL_PATH = "/dev/rfkill"
class BluetoothState(enum.Enum):
UNAVAILABLE = "unavailable"
ACTIVE = "active"
INACTIVE = "inactive"
def _unpack_variant(v):
if v is None:
@@ -38,6 +48,23 @@ def _make_proxy(bus: Gio.DBusConnection, path: str, iface: str) -> Gio.DBusProxy
)
def _rfkill_soft_blocked() -> bool:
try:
result = subprocess.run(
["rfkill", "list", "bluetooth"],
capture_output=True,
text=True,
timeout=5,
)
return "Soft blocked: yes" in result.stdout
except FileNotFoundError:
logger.warning("[Bluetooth] rfkill not found, assuming not soft blocked")
return False
except Exception as e:
logger.warning(f"[Bluetooth] rfkill check failed: {e}")
return False
class BluetoothDevice(Service):
@Signal
def changed(self) -> None: ...
@@ -570,10 +597,14 @@ class BluetoothClient(Service):
@Property(str, "readable")
def state(self) -> str:
if not self._adapters:
return BluetoothState.UNAVAILABLE.value
if _rfkill_soft_blocked():
return BluetoothState.INACTIVE.value
for a in self._adapters.values():
if a.powered:
return a.state
return "absent" if not self._adapters else "off"
return BluetoothState.INACTIVE.value
@Property(bool, "read-write", default_value=False)
def enabled(self) -> bool:
@@ -639,8 +670,19 @@ class BluetoothClient(Service):
None,
)
self._cached_rfkill_blocked = _rfkill_soft_blocked()
self._rfkill_timeout_id = GLib.timeout_add_seconds(10, self._check_rfkill)
self._populate_from_object_manager()
def _check_rfkill(self) -> bool:
blocked = _rfkill_soft_blocked()
if blocked != self._cached_rfkill_blocked:
self._cached_rfkill_blocked = blocked
self.notify("state")
self.emit("changed")
return True
def _populate_from_object_manager(self):
try:
result = self._om_proxy.call_sync(
+1 -1
View File
@@ -375,7 +375,7 @@ class PlayerService(Service):
if v is not None:
dur = (
v.get_uint64() / 1_000_000
if v.get_type_string() == "t"
if isinstance(v, GLib.Variant) and v.get_type_string() == "t"
else int(v) / 1_000_000
)
self._handle_artwork(metadata)