Skip to content

Instantly share code, notes, and snippets.

@infirit
Created November 12, 2022 19:54
Show Gist options
  • Save infirit/b9409c9e8108535b374f3cd6bf2ea72e to your computer and use it in GitHub Desktop.
Save infirit/b9409c9e8108535b374f3cd6bf2ea72e to your computer and use it in GitHub Desktop.
Fix invalidated properties
diff --git a/blueman/bluez/Base.py b/blueman/bluez/Base.py
index 6e6af0de..ae997426 100644
--- a/blueman/bluez/Base.py
+++ b/blueman/bluez/Base.py
@@ -1,14 +1,27 @@
from typing import List, Callable, Optional, Any, Union, Dict
from gi.repository import Gio, GLib, GObject
-from gi.types import GObjectMeta
from blueman.bluez.errors import parse_dbus_error, BluezDBusException
import logging
from blueman.bluemantyping import GSignals
-class BaseMeta(GObjectMeta):
+class Base(GObject.Object):
+ __name = 'org.bluez'
+ __bus_type = Gio.BusType.SYSTEM
+ __proxy = Gio.DBusProxy
+
+ __gsignals__: GSignals = {
+ 'property-changed': (GObject.SignalFlags.NO_HOOKS, None, (str, object, str))
+ }
+ __instances__: Dict[str, "Base"]
+
+ _interface_name: str
+
+ connect_signal = GObject.GObject.connect
+ disconnect_signal = GObject.GObject.disconnect
+
def __call__(cls, *args: object, **kwargs: str) -> "Base":
if not hasattr(cls, "__instances__"):
cls.__instances__: Dict[str, "Base"] = {}
@@ -25,54 +38,47 @@ class BaseMeta(GObjectMeta):
return instance
+ def __init__(self, *, obj_path: str):
+ super().__init__()
-class Base(Gio.DBusProxy, metaclass=BaseMeta):
- connect_signal = GObject.GObject.connect
- disconnect_signal = GObject.GObject.disconnect
-
- __name = 'org.bluez'
- __bus_type = Gio.BusType.SYSTEM
+ self.__proxy = Gio.DBusProxy.new_for_bus_sync(
+ self.__bus_type,
+ Gio.DBusProxyFlags.NONE,
+ None,
+ self.__name,
+ obj_path,
+ self._interface_name,
+ None
+ )
- __gsignals__: GSignals = {
- 'property-changed': (GObject.SignalFlags.NO_HOOKS, None, (str, object, str))
- }
- __instances__: Dict[str, "Base"]
-
- _interface_name: str
+ self.__proxy.connect("g-properties-changed", self.__properties_changed)
- def __init__(self, *, obj_path: str):
- super().__init__(
- g_name=self.__name,
- g_interface_name=self._interface_name,
- g_object_path=obj_path,
- g_bus_type=self.__bus_type,
- # FIXME See issue 620
- g_flags=Gio.DBusProxyFlags.GET_INVALIDATED_PROPERTIES)
-
- self.init()
self.__fallback = {'Icon': 'blueman', 'Class': 0, 'Appearance': 0}
self.__variant_map = {str: 's', int: 'u', bool: 'b'}
- def do_g_properties_changed(self, changed_properties: GLib.Variant, _invalidated_properties: List[str]) -> None:
+ def __properties_changed(self, _proxy: Gio.DBusProxy, changed_properties: GLib.Variant,
+ invalidated_properties: List[str]) -> None:
changed = changed_properties.unpack()
object_path = self.get_object_path()
- logging.debug(f"{object_path} {changed}")
+ logging.debug(f"{object_path} {changed} {invalidated_properties} {self}")
for key, value in changed.items():
self.emit("property-changed", key, value, object_path)
+ for key in invalidated_properties:
+ self.emit("property-changed", key, None, object_path)
def _call(
- self,
- method: str,
- param: Optional[GLib.Variant] = None,
- reply_handler: Optional[Callable[..., None]] = None,
- error_handler: Optional[Callable[[BluezDBusException], None]] = None,
+ self,
+ method: str,
+ param: Optional[GLib.Variant] = None,
+ reply_handler: Optional[Callable[..., None]] = None,
+ error_handler: Optional[Callable[[BluezDBusException], None]] = None,
) -> None:
def callback(
- proxy: Base,
- result: Gio.AsyncResult,
- reply: Optional[Callable[..., None]],
- error: Optional[Callable[[BluezDBusException], None]],
+ proxy: Gio.DBusProxy,
+ result: Gio.AsyncResult,
+ reply: Optional[Callable[..., None]],
+ error: Optional[Callable[[BluezDBusException], None]],
) -> None:
try:
value = proxy.call_finish(result).unpack()
@@ -82,14 +88,14 @@ class Base(Gio.DBusProxy, metaclass=BaseMeta):
if error:
error(parse_dbus_error(e))
else:
- logging.error(f"Unhandled error for {self.get_interface_name()}.{method}", exc_info=True)
+ logging.error(f"Unhandled error for {self.__proxy.get_interface_name()}.{method}", exc_info=True)
- self.call(method, param, Gio.DBusCallFlags.NONE, GLib.MAXINT, None,
- callback, reply_handler, error_handler)
+ self.__proxy.call(method, param, Gio.DBusCallFlags.NONE, GLib.MAXINT, None,
+ callback, reply_handler, error_handler)
def get(self, name: str) -> Any:
try:
- prop = self.call_sync(
+ prop = self.__proxy.call_sync(
'org.freedesktop.DBus.Properties.Get',
GLib.Variant('(ss)', (self._interface_name, name)),
Gio.DBusCallFlags.NONE,
@@ -97,7 +103,7 @@ class Base(Gio.DBusProxy, metaclass=BaseMeta):
None)
return prop.unpack()[0]
except GLib.Error as e:
- property = self.get_cached_property(name)
+ property = self.__proxy.get_cached_property(name)
if property is not None:
return property.unpack()
elif name in self.__fallback:
@@ -108,19 +114,22 @@ class Base(Gio.DBusProxy, metaclass=BaseMeta):
def set(self, name: str, value: Union[str, int, bool]) -> None:
v = GLib.Variant(self.__variant_map[type(value)], value)
param = GLib.Variant('(ssv)', (self._interface_name, name, v))
- self.call('org.freedesktop.DBus.Properties.Set',
- param,
- Gio.DBusCallFlags.NONE,
- GLib.MAXINT,
- None)
+ self.__proxy.call('org.freedesktop.DBus.Properties.Set',
+ param,
+ Gio.DBusCallFlags.NONE,
+ GLib.MAXINT,
+ None)
+
+ def get_object_path(self):
+ return self.__proxy.get_object_path()
def get_properties(self) -> Dict[str, Any]:
param = GLib.Variant('(s)', (self._interface_name,))
- res = self.call_sync('org.freedesktop.DBus.Properties.GetAll',
- param,
- Gio.DBusCallFlags.NONE,
- GLib.MAXINT,
- None)
+ res = self.__proxy.call_sync('org.freedesktop.DBus.Properties.GetAll',
+ param,
+ Gio.DBusCallFlags.NONE,
+ GLib.MAXINT,
+ None)
props: Dict[str, Any] = res.unpack()[0]
for k, v in self.__fallback.items():
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment