Daniel Brötzmann pushed to branch master at gajim / gajim

Commits: b21320e6 by wurstsalat at 2022-06-06T18:15:58+02:00 chore: Add type annotations to DBus modules

- - - - -

4 changed files:

- gajim/common/dbus/location.py - gajim/common/dbus/logind.py - gajim/common/dbus/music_track.py - gajim/common/dbus/system_style.py

Changes:

===================================== gajim/common/dbus/location.py ===================================== @@ -14,12 +14,18 @@ # You should have received a copy of the GNU General Public License # along with Gajim. If not, see http://www.gnu.org/licenses/.

+from __future__ import annotations + +from typing import Any from typing import Optional

import logging from datetime import datetime

+from gi.repository import Gio from gi.repository import GLib +from gi.repository import GObject + from nbxmpp.structs import LocationData

from gajim.common import app @@ -34,21 +40,21 @@

class LocationListener: - _instance = None + _instance: Optional[LocationListener] = None

@classmethod - def get(cls): + def get(cls) -> LocationListener: if cls._instance is None: cls._instance = cls() if app.is_installed('GEOCLUE') and not cls._instance._running: cls._instance.start() return cls._instance

- def __init__(self): - self._data = {} - self.location_info = {} - self.simple = None - self._current_location = None + def __init__(self) -> None: + self._data: dict[str, Any] = {} + self.location_info: dict[str, Any] = {} + self.simple: Optional[Geoclue.Simple] = None + self._current_location: Optional[LocationData] = None self._running = False

def _emit(self, info: Optional[LocationData]) -> None: @@ -59,33 +65,38 @@ def _emit(self, info: Optional[LocationData]) -> None: def current_location(self) -> Optional[LocationData]: return self._current_location

- # Note: do not remove third parameter `param` - # because notify signal expects three parameters - def _on_location_update(self, simple, _param=None): + def _on_location_update(self, simple: Geoclue.Simple, *args: Any) -> None: location = simple.get_location() - timestamp = location.get_property("timestamp")[0] - lat = location.get_property("latitude") - lon = location.get_property("longitude") - alt = location.get_property("altitude") + timestamp = location.get_property('timestamp')[0] + lat = location.get_property('latitude') + lon = location.get_property('longitude') + alt = location.get_property('altitude') # in XEP-0080 it's horizontal accuracy - acc = location.get_property("accuracy") + acc = location.get_property('accuracy')

# update data with info we just received self._data = {'lat': lat, 'lon': lon, 'alt': alt, 'accuracy': acc} self._data['timestamp'] = self._timestamp_to_utc(timestamp) self._send_location()

- def _on_client_update(self, client, _param=None): + def _on_client_update(self, + client: Geoclue.Client, + *args: Any + ) -> None: if not client.get_property('active'): self._emit(None)

- def _on_simple_ready(self, _obj, result): + def _on_simple_ready(self, + _obj: GObject.Object, + result: Gio.AsyncResult + ) -> None: try: self.simple = Geoclue.Simple.new_finish(result) except GLib.Error as error: - log.warning("Could not enable geolocation: %s", error.message) + log.warning('Could not enable geolocation: %s', error.message) self._running = False else: + assert self.simple is not None self.simple.connect('notify::location', self._on_location_update) client = self.simple.get_client() # Inside the flatpak sandbox client will be None, @@ -95,18 +106,18 @@ def _on_simple_ready(self, _obj, result):

self._on_location_update(self.simple)

- def get_data(self): - Geoclue.Simple.new("org.gajim.Gajim", + def get_data(self) -> None: + Geoclue.Simple.new('org.gajim.Gajim', Geoclue.AccuracyLevel.EXACT, None, self._on_simple_ready)

- def start(self): + def start(self) -> None: self._running = True self.location_info = {} self.get_data()

- def _send_location(self): + def _send_location(self) -> None: if self.location_info == self._data: return if 'timestamp' in self.location_info and 'timestamp' in self._data: @@ -121,6 +132,6 @@ def _send_location(self): self._emit(info)

@staticmethod - def _timestamp_to_utc(timestamp): + def _timestamp_to_utc(timestamp: float) -> str: time = datetime.utcfromtimestamp(timestamp) return time.strftime('%Y-%m-%dT%H:%MZ')

===================================== gajim/common/dbus/logind.py ===================================== @@ -18,6 +18,10 @@ Watch for system sleep using systemd-logind. Documentation: http://www.freedesktop.org/wiki/Software/systemd/inhibit ''' +from __future__ import annotations + +from typing import Any +from typing import Optional

import os import logging @@ -32,17 +36,17 @@

class LogindListener: - _instance = None + _instance: Optional[LogindListener] = None

@classmethod - def get(cls): + def get(cls) -> LogindListener: if cls._instance is None: cls._instance = cls() return cls._instance

- def __init__(self): + def __init__(self) -> None: # file descriptor object of the inhibitor - self._inhibit_fd = None + self._inhibit_fd: Optional[int] = None

Gio.bus_watch_name( Gio.BusType.SYSTEM, @@ -51,14 +55,20 @@ def __init__(self): self._on_appear_logind, self._on_vanish_logind)

- def _on_prepare_for_sleep(self, connection, _sender_name, _object_path, - interface_name, signal_name, parameters, - *_user_data): + def _on_prepare_for_sleep(self, + connection: Gio.DBusConnection, + _sender_name: str, + _object_path: str, + interface_name: str, + signal_name: str, + parameters: tuple[str, str, str], + *_user_data: Any + ) -> None: '''Signal handler for PrepareForSleep event''' log.debug('Received signal %s.%s%s', interface_name, signal_name, parameters)

- before = parameters[0] # Signal is either before or after sleep occurs + before = parameters[0] # Signal is either before or after sleep occurs if before: warn = self._inhibit_fd is None log.log( @@ -86,7 +96,7 @@ def _on_prepare_for_sleep(self, connection, _sender_name, _object_path, if conn.state.is_disconnected and conn.time_to_reconnect: conn.reconnect()

- def _inhibit_sleep(self, connection): + def _inhibit_sleep(self, connection: Gio.DBusConnection) -> None: '''Obtain a sleep delay inhibitor from logind''' if self._inhibit_fd is not None: # Something is wrong, we have an inhibitor fd, and we are asking for @@ -95,7 +105,7 @@ def _inhibit_sleep(self, connection): 'while already holding one.')

try: - ret, ret_fdlist = connection.call_with_unix_fd_list_sync( + result = connection.call_with_unix_fd_list_sync( 'org.freedesktop.login1', '/org/freedesktop/login1', 'org.freedesktop.login1.Manager', @@ -104,10 +114,18 @@ def _inhibit_sleep(self, connection): 'sleep', 'org.gajim.Gajim', _('Disconnect from the network'), - 'delay' # Inhibitor will delay but not block sleep + 'delay' # Inhibitor will delay but not block sleep )), GLib.VariantType.new('(h)'), - Gio.DBusCallFlags.NONE, -1, None, None) + Gio.DBusCallFlags.NONE, + -1, + None, + None) + if result is None: + log.warning( + 'Cannot obtain a sleep delay inhibitor from logind') + return + ret, ret_fdlist = result except GLib.Error as error: log.warning( 'Cannot obtain a sleep delay inhibitor from logind: %s', error) @@ -116,14 +134,19 @@ def _inhibit_sleep(self, connection): log.info('Inhibit sleep') self._inhibit_fd = ret_fdlist.get(ret.unpack()[0])

- def _disinhibit_sleep(self): + def _disinhibit_sleep(self) -> None: '''Relinquish our sleep delay inhibitor''' if self._inhibit_fd is not None: os.close(self._inhibit_fd) self._inhibit_fd = None log.info('Disinhibit sleep')

- def _on_appear_logind(self, connection, name, name_owner, *_user_data): + def _on_appear_logind(self, + connection: Gio.DBusConnection, + name: str, + name_owner: str, + *_user_data: Any + ) -> None: '''Use signal and locks provided by org.freedesktop.login1''' log.info('Name %s appeared, owned by %s', name, name_owner)

@@ -138,13 +161,17 @@ def _on_appear_logind(self, connection, name, name_owner, *_user_data): None) self._inhibit_sleep(connection)

- def _on_vanish_logind(self, _connection, name, *_user_data): + def _on_vanish_logind(self, + _connection: Gio.DBusConnection, + name: str, + *_user_data: Any + ) -> None: '''Release remaining resources related to org.freedesktop.login1''' log.info('Name %s vanished', name) self._disinhibit_sleep()

-def enable(): +def enable() -> None: # Module should be rewritten # Its not useful to know when the computer goes to sleep. # Because of stream management we don’t need to do anything in that case.

===================================== gajim/common/dbus/music_track.py ===================================== @@ -19,6 +19,10 @@ # You should have received a copy of the GNU General Public License # along with Gajim. If not, see http://www.gnu.org/licenses/.

+from __future__ import annotations + +from typing import Any +from typing import cast from typing import Optional

import logging @@ -38,18 +42,18 @@

class MusicTrackListener:

- _instance = None + _instance: Optional[MusicTrackListener] = None

@classmethod - def get(cls): + def get(cls) -> MusicTrackListener: if cls._instance is None: cls._instance = cls() return cls._instance

- def __init__(self): - self.players = {} - self.connection = None - self._current_tune = None + def __init__(self) -> None: + self.players: dict[str, int] = {} + self.connection: Optional[Gio.DBusConnection] = None + self._current_tune: Optional[TuneData] = None self._running = False

if sys.platform not in ('win32', 'darwin'): @@ -94,7 +98,7 @@ def start(self): -1, None) except GLib.Error as error: - log.debug("Could not list names: %s", error.message) + log.debug('Could not list names: %s', error.message) return

for name in result[0]: @@ -106,19 +110,19 @@ def start(self):

self._running = True

- def stop(self): + def stop(self) -> None: for name in list(self.players): if name.startswith(MPRIS_PLAYER_PREFIX): self._remove_player(name)

def _signal_name_owner_changed(self, - _connection, - _sender_name, - _object_path, - _interface_name, - _signal_name, - parameters, - *_user_data): + _connection: Gio.DBusConnection, + _sender_name: str, + _object_path: str, + _interface_name: str, + _signal_name: str, + parameters: tuple[str, str, str], + *_user_data: Any): name, old_owner, new_owner = parameters if name.startswith(MPRIS_PLAYER_PREFIX): if new_owner and not old_owner: @@ -126,13 +130,14 @@ def _signal_name_owner_changed(self, else: self._remove_player(name)

- def _add_player(self, name): + def _add_player(self, name: str) -> None: '''Set up a listener for music player signals''' log.info('%s appeared', name)

if name in self.players: return

+ assert self.connection is not None self.players[name] = self.connection.signal_subscribe( name, 'org.freedesktop.DBus.Properties', @@ -143,9 +148,13 @@ def _add_player(self, name): self._signal_received, name)

- def _remove_player(self, name): + def _remove_player(self, name: str) -> None: log.info('%s vanished', name) if name in self.players: + if self.connection is None: + self._emit(None) + return + self.connection.signal_unsubscribe( self.players[name]) self.players.pop(name) @@ -153,21 +162,21 @@ def _remove_player(self, name): self._emit(None)

def _signal_received(self, - _connection, - _sender_name, - _object_path, - interface_name, - _signal_name, - parameters, - *user_data): + _connection: Gio.DBusConnection, + _sender_name: str, + _object_path: str, + interface_name: str, + _signal_name: str, + parameters: tuple[str, str, str], + *user_data: str): '''Signal handler for PropertiesChanged event'''

log.info('Signal received: %s - %s', interface_name, parameters) self._get_playing_track(user_data[0])

@staticmethod - def _get_music_info(properties): - meta = properties.get('Metadata') + def _get_music_info(properties: dict[str, str]) -> Optional[TuneData]: + meta = cast(dict[str, str], properties.get('Metadata')) if meta is None or not meta: return None

@@ -183,7 +192,7 @@ def _get_music_info(properties): artist = ', '.join(artist) return TuneData(artist=artist, title=title, source=album)

- def _get_playing_track(self, name): + def _get_playing_track(self, name: str) -> None: '''Return a TuneData for the currently playing song, or None if no song is playing''' proxy = Gio.DBusProxy.new_for_bus_sync( @@ -195,18 +204,21 @@ def _get_playing_track(self, name): 'org.freedesktop.DBus.Properties', None)

- def proxy_call_finished(proxy, res): + def proxy_call_finished(proxy: Gio.DBusProxy, + res: Gio.AsyncResult + ) -> None: try: result = proxy.call_finish(res) except GLib.Error as error: - log.debug("Could not enable music listener: %s", error.message) + log.debug('Could not enable music listener: %s', error.message) return

+ assert result is not None info = self._get_music_info(result[0]) if info is not None: self._emit(info)

- proxy.call("GetAll", + proxy.call('GetAll', GLib.Variant('(s)', ('org.mpris.MediaPlayer2.Player',)), Gio.DBusCallFlags.NONE, -1,

===================================== gajim/common/dbus/system_style.py ===================================== @@ -13,7 +13,11 @@ # along with Gajim. If not, see http://www.gnu.org/licenses/.

from __future__ import annotations + +from typing import Any +from typing import Callable from typing import cast +from typing import Optional

import sys import logging @@ -28,8 +32,8 @@

class SystemStyleListener: - def __init__(self, callback) -> None: - self._prefer_dark = None + def __init__(self, callback: Callable[..., None]) -> None: + self._prefer_dark: Optional[bool] = None self._callback = callback

if sys.platform in ('win32', 'darwin'): @@ -57,32 +61,35 @@ def read_color_scheme(self) -> None: result = self.dbus_proxy.call_sync( 'Read', GLib.Variant('(ss)', ('org.freedesktop.appearance', - 'color-scheme')), + 'color-scheme')), Gio.DBusCallFlags.NO_AUTO_START, -1, None) (self._prefer_dark,) = cast(tuple[bool], result) except GLib.Error as error: - log.error("Couldn't read the color-scheme setting: %s", + log.error('Couldn’t read the color-scheme setting: %s', error.message) + self._prefer_dark = False return

def _signal_setting_changed(self, - _proxy, - _sender_name, - signal_name, - parameters, - *_user_data) -> None: + _proxy: Gio.DBusProxy, + _sender_name: Optional[str], + signal_name: str, + parameters: GLib.Variant, + *_user_data: Any + ) -> None: if signal_name != 'SettingChanged': return

namespace, name, value = parameters - if (namespace == "org.freedesktop.appearance" and - name == "color-scheme"): + if (namespace == 'org.freedesktop.appearance' and + name == 'color-scheme'): self._prefer_dark = (value == 1) self._callback() app.ged.raise_event(StyleChanged())

@property def prefer_dark(self) -> bool: + assert self._prefer_dark is not None return self._prefer_dark

View it on GitLab: https://dev.gajim.org/gajim/gajim/-/commit/b21320e603877b53b1bf9a4654d8aec2f...