Add stubs/gi

blueman/bluez/AnyBase.py

@@ -1,4 +1,4 @@
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 from gi.repository import GObject, GLib
 from gi.repository import Gio
@@ -20,7 +20,7 @@ class AnyBase(GObject.GObject):
     def __init__(self, interface_name: str):
         super().__init__()
 
-        self.__bus = Gio.bus_get_sync(Gio.BusType.SYSTEM)
+        self.__bus: Optional[Gio.DBusConnection] = Gio.bus_get_sync(Gio.BusType.SYSTEM)
 
         self.__interface_name = interface_name
         self.__signal = None
@@ -49,6 +49,7 @@ def _on_properties_changed(
 
     def close(self) -> None:
         if self.__signal:
-            self.__bus.signal_unsubscribe(self.__signal)
+            if self.__bus is not None:
+                self.__bus.signal_unsubscribe(self.__signal)
             self.__signal = None
         self.__bus = None

blueman/bluez/Base.py

@@ -38,6 +38,8 @@ class Base(Gio.DBusProxy, metaclass=BaseMeta):
     }
     __instances__: Dict[str, "Base"]
 
+    _interface_name: str
+
     def __init__(self, obj_path: str):
         super().__init__(
             g_name=self.__name,
@@ -68,7 +70,7 @@ def _call(
     ) -> None:
         def callback(
             proxy: Base,
-            result: Gio.Task,
+            result: Gio.AsyncResult,
             reply: Optional[Callable[..., None]],
             error: Optional[Callable[[BluezDBusException], None]],
         ) -> None:
@@ -95,8 +97,9 @@ def get(self, name: str) -> Any:
                 None)
             return prop.unpack()[0]
         except GLib.Error as e:
-            if name in self.get_cached_property_names():
-                return self.get_cached_property(name).unpack()
+            property = self.get_cached_property(name)
+            if property is not None:
+                return property.unpack()
             elif name in self.__fallback:
                 return self.__fallback[name]
             else:

blueman/bluez/Device.py

@@ -18,14 +18,14 @@ def pair(
     ) -> None:
         self._call('Pair', reply_handler=reply_handler, error_handler=error_handler)
 
-    def connect(
+    def connect(  # type: ignore
         self,
         reply_handler: Optional[Callable[[], None]] = None,
         error_handler: Optional[Callable[[BluezDBusException], None]] = None,
     ) -> None:
         self._call('Connect', reply_handler=reply_handler, error_handler=error_handler)
 
-    def disconnect(
+    def disconnect(  # type: ignore
         self,
         reply_handler: Optional[Callable[[], None]] = None,
         error_handler: Optional[Callable[[BluezDBusException], None]] = None,

blueman/bluez/Manager.py

@@ -37,10 +37,12 @@ def _on_object_added(self, _object_manager: Gio.DBusObjectManager, dbus_object:
         adapter_proxy = dbus_object.get_interface('org.bluez.Adapter1')
 
         if adapter_proxy:
+            assert isinstance(adapter_proxy, Gio.DBusProxy)
             object_path = adapter_proxy.get_object_path()
             logging.debug(object_path)
             self.emit('adapter-added', object_path)
         elif device_proxy:
+            assert isinstance(device_proxy, Gio.DBusProxy)
             object_path = device_proxy.get_object_path()
             logging.debug(object_path)
             self.emit('device-created', object_path)
@@ -50,10 +52,12 @@ def _on_object_removed(self, _object_manager: Gio.DBusObjectManager, dbus_object
         adapter_proxy = dbus_object.get_interface('org.bluez.Adapter1')
 
         if adapter_proxy:
+            assert isinstance(adapter_proxy, Gio.DBusProxy)
             object_path = adapter_proxy.get_object_path()
             logging.debug(object_path)
             self.emit('adapter-removed', object_path)
         elif device_proxy:
+            assert isinstance(device_proxy, Gio.DBusProxy)
             object_path = device_proxy.get_object_path()
             logging.debug(object_path)
             self.emit('device-removed', object_path)
@@ -64,6 +68,7 @@ def get_adapters(self) -> List[Adapter]:
             proxy = obj_proxy.get_interface('org.bluez.Adapter1')
 
             if proxy:
+                assert isinstance(proxy, Gio.DBusProxy)
                 paths.append(proxy.get_object_path())
 
         return [Adapter(obj_path=path) for path in paths]
@@ -88,6 +93,7 @@ def get_devices(self, adapter_path: str = "/") -> List[Device]:
             proxy = obj_proxy.get_interface('org.bluez.Device1')
 
             if proxy:
+                assert isinstance(proxy, Gio.DBusProxy)
                 object_path = proxy.get_object_path()
                 if object_path.startswith(adapter_path):
                     paths.append(object_path)

blueman/bluez/Network.py

@@ -13,7 +13,7 @@ class Network(Base):
     def __init__(self, obj_path: str):
         super().__init__(obj_path=obj_path)
 
-    def connect(
+    def connect(  # type: ignore
         self,
         uuid: str,
         reply_handler: Optional[Callable[[str], None]] = None,
@@ -22,7 +22,7 @@ def connect(
         param = GLib.Variant('(s)', (uuid,))
         self._call('Connect', param, reply_handler=reply_handler, error_handler=error_handler)
 
-    def disconnect(
+    def disconnect(  # type: ignore
         self,
         reply_handler: Optional[Callable[[], None]] = None,
         error_handler: Optional[Callable[[BluezDBusException], None]] = None,

blueman/gui/DeviceList.py

@@ -72,7 +72,7 @@ def __init__(self, adapter_name: Optional[str] = None, tabledata: Optional[List[
         self._anydevhandler = self.any_device.connect_signal("property-changed", self._on_device_property_changed)
 
         self.__discovery_time: float = 0
-        self.__adapter_path = None
+        self.__adapter_path: Optional[str] = None
         self.Adapter: Optional[Adapter] = None
         self.discovering = False
 

blueman/gui/Notification.py

@@ -249,7 +249,7 @@ def show(self) -> None:
 
     def close(self) -> None:
         param = GLib.Variant('(u)', (self._return_id,))
-        self.call_sync('CloseNotification', param, Gio.DBusProxyFlags.NONE, -1, None)
+        self.call_sync('CloseNotification', param, Gio.DBusCallFlags.NONE, -1, None)
         self._return_id = None
 
 

blueman/main/DBusProxies.py

@@ -9,7 +9,7 @@ class DBusProxyFailed(Exception):
 
 class ProxyBase(Gio.DBusProxy, metaclass=SingletonGObjectMeta):
     def __init__(self, name: str, interface_name: str, object_path: str = "/", systembus: bool = False,
-                 flags: Gio.DBusProxyFlags = 0) -> None:
+                 flags: Gio.DBusProxyFlags = Gio.DBusProxyFlags.NONE) -> None:
         if systembus:
             bustype = Gio.BusType.SYSTEM
         else:
@@ -52,7 +52,7 @@ def call_finish(proxy: "ManagerService", resp: Gio.AsyncResult) -> None:
             proxy.call_finish(resp)
 
         param = GLib.Variant('(sava{sv})', (name, [], {}))
-        self.call('ActivateAction', param, Gio.DBusProxyFlags.NONE, -1, None, call_finish)
+        self.call('ActivateAction', param, Gio.DBusCallFlags.NONE, -1, None, call_finish)
 
     def startstop(self) -> None:
         if self.get_name_owner() is None:

blueman/main/DbusService.py

@@ -9,15 +9,15 @@
 class DbusError(Exception):
     _name = "org.blueman.Error"
 
-    def __init__(self, message: Optional[str] = None) -> None:
+    def __init__(self, message: str) -> None:
         self._message = message
 
     @property
     def name(self) -> str:
         return self._name
 
     @property
-    def message(self) -> Optional[str]:
+    def message(self) -> str:
         return self._message
 
 
@@ -30,7 +30,7 @@ def __init__(self, bus_name: Optional[str], interface_name: str, path: str, bus_
         self._signals: Dict[str, str] = {}
         self._interface_name = interface_name
         self._path = path
-        self._regid = None
+        self._regid: Optional[int] = None
 
     def add_method(self, name: str, arguments: Tuple[str, ...], return_value: str, method: Callable[..., None],
                    pass_sender: bool = False, is_async: bool = False) -> None:
@@ -95,8 +95,9 @@ def register(self) -> None:
             raise GLib.Error(f"Failed to register object with path: {self._path}")
 
     def unregister(self) -> None:
-        self._bus.unregister_object(self._regid)
-        self._regid = None
+        if self._regid is not None:
+            self._bus.unregister_object(self._regid)
+            self._regid = None
 
     def _reregister(self) -> None:
         if self._regid:

blueman/main/MechanismApplication.py

@@ -1,6 +1,7 @@
 import importlib
 import logging
 import os
+from typing import Optional
 
 import blueman.plugins.mechanism
 from blueman.Constants import POLKIT
@@ -47,7 +48,7 @@ def __init__(self, stoptimer: bool):
 
         if POLKIT:
             try:
-                self.pk = Gio.DBusProxy.new_for_bus_sync(
+                self.pk: Optional[Gio.DBusProxy] = Gio.DBusProxy.new_for_bus_sync(
                     Gio.BusType.SYSTEM,
                     Gio.DBusProxyFlags.NONE,
                     None,

blueman/main/Sendto.py

@@ -214,7 +214,9 @@ def on_transfer_completed(self, _transfer: Transfer) -> None:
 
     def process_queue(self) -> None:
         if len(self.files) > 0:
-            self.send_file(self.files[-1].get_path())
+            path = self.files[-1].get_path()
+            assert path is not None
+            self.send_file(path)
         else:
             self.emit("result", True)
 

blueman/main/applet/BluezAgent.py

@@ -295,8 +295,7 @@ def on_auth_action(action: str) -> None:
             logging.info(action)
 
             if action == "always":
-                device = Device(obj_path=n._device)
-                device.set("Trusted", True)
+                Device(obj_path=device).set("Trusted", True)
             if action == "always" or action == "accept":
                 ok()
             else:
@@ -313,4 +312,3 @@ def on_auth_action(action: str) -> None:
 
         n = Notification(_("Bluetooth Authentication"), notify_message, 0, actions, on_auth_action, icon_name="blueman")
         n.show()
-        n._device = device

blueman/plugins/applet/KillSwitch.py

@@ -1,6 +1,6 @@
 from gettext import gettext as _
 import os
-from typing import Dict, Callable, Any
+from typing import Dict, Callable, Any, Optional
 
 from gi.repository import GLib, Gio
 import struct
@@ -54,7 +54,7 @@ class KillSwitch(AppletPlugin, PowerStateHandler, StatusIconVisibilityHandler):
     _hardblocked = False
 
     def on_load(self) -> None:
-        self._connman_proxy = None
+        self._connman_proxy: Optional[Gio.DBusProxy] = None
         self._connman_watch_id = Gio.bus_watch_name(Gio.BusType.SYSTEM, "net.connman", Gio.BusNameWatcherFlags.NONE,
                                                     self._on_connman_appeared, self._on_connman_vanished)
 

blueman/plugins/applet/TransferService.py

@@ -230,7 +230,7 @@ def _unregister_agent(self) -> None:
             self._agent.unregister()
             self._agent = None
 
-    def _on_dbus_name_appeared(self, _connection: Manager, name: str, owner: str) -> None:
+    def _on_dbus_name_appeared(self, _connection: Gio.DBusConnection, name: str, owner: str) -> None:
         logging.info(f"{name} {owner}")
 
         self._manager = Manager()
@@ -240,7 +240,7 @@ def _on_dbus_name_appeared(self, _connection: Manager, name: str, owner: str) ->
 
         self._register_agent()
 
-    def _on_dbus_name_vanished(self, _connection: Manager, name: str) -> None:
+    def _on_dbus_name_vanished(self, _connection: Gio.DBusConnection, name: str) -> None:
         logging.info(f"{name} not running or was stopped")
 
         if self._manager:

blueman/services/meta/SerialService.py

@@ -44,7 +44,9 @@ def on_file_changed(
             else:
                 logging.warning(f"No handler id for {port}")
         elif event_type == Gio.FileMonitorEvent.ATTRIBUTE_CHANGED:
-            self.try_replace_root_watcher(monitor, file.get_path(), port)
+            path = file.get_path()
+            assert path is not None
+            self.try_replace_root_watcher(monitor, path, port)
 
     def try_replace_root_watcher(self, monitor: Gio.FileMonitor, path: str, port: int) -> None:
         if not os.access(path, os.R_OK | os.W_OK):