diff --git a/sublime/app.py b/sublime/app.py index 5deca30..ffa0b59 100644 --- a/sublime/app.py +++ b/sublime/app.py @@ -2,7 +2,8 @@ import logging import math import os import random -from typing import Any, Callable, Dict, Iterable, List, Optional +from concurrent.futures import Future +from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple import gi gi.require_version('Gtk', '3.0') @@ -24,7 +25,7 @@ class SublimeMusicApp(Gtk.Application): super().__init__(application_id="com.sumnerevans.sublimemusic") Notify.init('Sublime Music') - self.window = None + self.window: Optional[Gtk.Window] = None self.state = ApplicationState() self.state.config_file = config_file @@ -210,13 +211,13 @@ class SublimeMusicApp(Gtk.Application): def on_dbus_method_call( self, - connection, - sender, - path, - interface, - method, - params, - invocation, + connection: Gio.DBusConnection, + sender: str, + path: str, + interface: str, + method: str, + params: GLib.Variant, + invocation: Gio.DBusMethodInvocation, ): second_microsecond_conversion = 1000000 @@ -314,7 +315,7 @@ class SublimeMusicApp(Gtk.Application): reverse=reverse_order, ) - def make_playlist_tuple(p: Playlist): + def make_playlist_tuple(p: Playlist) -> GLib.Variant: cover_art_filename = CacheManager.get_cover_art_filename( p.coverArt, allow_download=False, @@ -323,15 +324,15 @@ class SublimeMusicApp(Gtk.Application): return GLib.Variant( '(a(oss))', ( - list( - map( - make_playlist_tuple, - playlists[index:(index + max_count)])), )) + [ + make_playlist_tuple(p) + for p in playlists[index:(index + max_count)] + ], )) - method_call_map = { + method_call_map: Dict[str, Dict[str, Any]] = { 'org.mpris.MediaPlayer2': { - 'Raise': self.window.present, - 'Quit': self.window.destroy, + 'Raise': self.window and self.window.present, + 'Quit': self.window and self.window.destroy, }, 'org.mpris.MediaPlayer2.Player': { 'Next': self.on_next_track, @@ -352,34 +353,35 @@ class SublimeMusicApp(Gtk.Application): 'GetPlaylists': get_playlists, }, } - method = method_call_map.get(interface, {}).get(method) - if method is None: + method_fn = method_call_map.get(interface, {}).get(method) + if method_fn is None: logging.warning( f'Unknown/unimplemented method: {interface}.{method}.') - invocation.return_value(method(*params) if callable(method) else None) + invocation.return_value( + method_fn(*params) if callable(method_fn) else None) def on_dbus_set_property( self, - connection, - sender, - path, - interface, - property_name, - value, + connection: Gio.DBusConnection, + sender: str, + path: str, + interface: str, + property_name: str, + value: GLib.Variant, ): - def change_loop(new_loop_status): + def change_loop(new_loop_status: GLib.Variant): self.state.repeat_type = RepeatType.from_mpris_loop_status( new_loop_status.get_string()) self.update_window() - def set_shuffle(new_val): + def set_shuffle(new_val: GLib.Variant): if new_val.get_boolean() != self.state.shuffle_on: self.on_shuffle_press(None, None) - def set_volume(new_val): - self.on_volume_change(None, value.get_double() * 100) + def set_volume(new_val: GLib.Variant): + self.on_volume_change(None, new_val.get_double() * 100) - setter_map = { + setter_map: Dict[str, Dict[str, Any]] = { 'org.mpris.MediaPlayer2.Player': { 'LoopStatus': change_loop, 'Rate': lambda _: None, @@ -388,7 +390,7 @@ class SublimeMusicApp(Gtk.Application): } } - setter = setter_map.get(interface).get(property_name) + setter = setter_map.get(interface, {}).get(property_name) if setter is None: logging.warning('Set: Unknown property: {property_name}.') return @@ -612,7 +614,13 @@ class SublimeMusicApp(Gtk.Application): self.state.current_tab = stack.get_visible_child_name() self.update_window() - def on_song_clicked(self, win, song_index, song_queue, metadata): + def on_song_clicked( + self, + win: Any, + song_index: int, + song_queue: List[str], + metadata: Dict[str, Any], + ): # Reset the play queue so that we don't ever revert back to the # previous one. old_play_queue = song_queue.copy() @@ -641,7 +649,7 @@ class SublimeMusicApp(Gtk.Application): play_queue=song_queue, ) - def on_songs_removed(self, win, song_indexes_to_remove): + def on_songs_removed(self, win: Any, song_indexes_to_remove: List[int]): self.state.play_queue = [ song_id for i, song_id in enumerate(self.state.play_queue) if i not in song_indexes_to_remove @@ -669,8 +677,8 @@ class SublimeMusicApp(Gtk.Application): self.save_play_queue() @dbus_propagate() - def on_song_scrub(self, _, scrub_value): - if not hasattr(self.state, 'current_song'): + def on_song_scrub(self, win: Any, scrub_value: float): + if not self.state.current_song or not self.window: return new_time = self.state.current_song.duration * (scrub_value / 100) @@ -685,7 +693,7 @@ class SublimeMusicApp(Gtk.Application): self.save_play_queue() - def on_device_update(self, _, device_uuid): + def on_device_update(self, win: Any, device_uuid: str): if device_uuid == self.state.current_device: return self.state.current_device = device_uuid @@ -709,24 +717,28 @@ class SublimeMusicApp(Gtk.Application): self.dbus_manager.property_diff() @dbus_propagate() - def on_mute_toggle(self, action, _): + def on_mute_toggle(self, *args): self.state.is_muted = not self.state.is_muted self.player.is_muted = self.state.is_muted self.update_window() @dbus_propagate() - def on_volume_change(self, _, value): + def on_volume_change(self, _: Any, value: float): self.state.volume = value self.player.volume = self.state.volume self.update_window() - def on_window_key_press(self, window, event): + def on_window_key_press( + self, + window: Gtk.Window, + event: Gdk.EventKey, + ) -> bool: # Need to use bitwise & here to see if CTRL is pressed. if (event.keyval == 102 and event.state & Gdk.ModifierType.CONTROL_MASK): # Ctrl + F window.search_entry.grab_focus() - return + return False if window.search_entry.has_focus(): return False @@ -742,7 +754,9 @@ class SublimeMusicApp(Gtk.Application): action() return True - def on_app_shutdown(self, app): + return False + + def on_app_shutdown(self, app: 'SublimeMusicApp'): Notify.uninit() if self.state.config.server is None: @@ -767,10 +781,12 @@ class SublimeMusicApp(Gtk.Application): dialog.run() dialog.destroy() - def update_window(self, force=False): + def update_window(self, force: bool = False): + if not self.window: + return GLib.idle_add(lambda: self.window.update(self.state, force=force)) - def update_play_state_from_server(self, prompt_confirm=False): + def update_play_state_from_server(self, prompt_confirm: bool = False): # TODO (#129): need to make the up next list loading for the duration # here if prompt_confirm is False. was_playing = self.state.playing @@ -778,7 +794,7 @@ class SublimeMusicApp(Gtk.Application): self.state.playing = False self.update_window() - def do_update(f): + def do_update(f: Future): play_queue = f.result() new_play_queue = [s.id for s in play_queue.entry] new_current_song_id = str(play_queue.current) @@ -835,9 +851,9 @@ class SublimeMusicApp(Gtk.Application): def play_song( self, song_index: int, - reset=False, - old_play_queue=None, - play_queue=None, + reset: bool = False, + old_play_queue: List[str] = None, + play_queue: List[str] = None, ): # Do this the old fashioned way so that we can have access to ``reset`` # in the callback. @@ -869,13 +885,14 @@ class SublimeMusicApp(Gtk.Application): song_notification.add_action( 'clicked', 'Open Sublime Music', - lambda *a: self.window.present(), + lambda *a: self.window.present() + if self.window else None, ) song_notification.show() def on_cover_art_download_complete( - cover_art_filename, - order_token, + cover_art_filename: str, + order_token: int, ): if order_token != self.song_playing_order_token: return @@ -887,7 +904,8 @@ class SublimeMusicApp(Gtk.Application): cover_art_filename, 70, 70, True)) song_notification.show() - def get_cover_art_filename(order_token): + def get_cover_art_filename( + order_token: int) -> Tuple[str, int]: return ( CacheManager.get_cover_art_filename( song.coverArt).result(), @@ -906,8 +924,9 @@ class SublimeMusicApp(Gtk.Application): 'Unable to display notification. Is a notification ' 'daemon running?') - def on_song_download_complete(song_id): - if self.state.current_song.id != song.id: + def on_song_download_complete(song_id: int): + if (self.state.current_song + and self.state.current_song.id != song.id): return # Switch to the local media if the player can hotswap (MPV can, diff --git a/sublime/cache_manager.py b/sublime/cache_manager.py index 4fd3777..fd3c796 100644 --- a/sublime/cache_manager.py +++ b/sublime/cache_manager.py @@ -416,32 +416,37 @@ class CacheManager(metaclass=Singleton): f.write(data) def calculate_abs_path(self, *relative_paths) -> Path: + server_hash = CacheManager.calculate_server_hash( + self.server_config) + if not server_hash: + raise Exception( + "Could not calculate the current server's hash.") return Path(self.app_config.cache_location).joinpath( - CacheManager.calculate_server_hash(self.server_config), - *relative_paths, - ) + server_hash, *relative_paths) def calculate_download_path(self, *relative_paths) -> Path: """ Determine where to temporarily put the file as it is downloading. """ + server_hash = CacheManager.calculate_server_hash( + self.server_config) + if not server_hash: + raise Exception( + "Could not calculate the current server's hash.") xdg_cache_home = ( os.environ.get('XDG_CACHE_HOME') or os.path.expanduser('~/.cache')) return Path(xdg_cache_home).joinpath( - 'sublime-music', - CacheManager.calculate_server_hash(self.server_config), - *relative_paths, - ) + 'sublime-music', server_hash, *relative_paths) def return_cached_or_download( - self, - relative_path: Union[Path, str], - download_fn: Callable[[], bytes], - before_download: Callable[[], None] = lambda: None, - force: bool = False, - allow_download: bool = True, - ) -> 'CacheManager.Result[Optional[str]]': + self, + relative_path: Union[Path, str], + download_fn: Callable[[], bytes], + before_download: Callable[[], None] = lambda: None, + force: bool = False, + allow_download: bool = True, + ) -> 'CacheManager.Result[str]': abs_path = self.calculate_abs_path(relative_path) abs_path_str = str(abs_path) download_path = self.calculate_download_path(relative_path) @@ -575,7 +580,7 @@ class CacheManager(metaclass=Singleton): return CacheManager.create_future(do_create_playlist) - def update_playlist(self, playlist_id, *args, **kwargs) -> Future: + def update_playlist(self, playlist_id: int, *args, **kwargs) -> Future: def do_update_playlist(): self.server.update_playlist(playlist_id, *args, **kwargs) with self.cache_lock: @@ -662,7 +667,7 @@ class CacheManager(metaclass=Singleton): def get_music_directory( self, - id, + id: int, before_download: Callable[[], None] = lambda: None, force: bool = False, ) -> 'CacheManager.Result[Directory]': @@ -715,10 +720,9 @@ class CacheManager(metaclass=Singleton): artist: Union[Artist, ArtistID3], before_download: Callable[[], None] = lambda: None, force: bool = False, - ) -> 'CacheManager.Result[Optional[str]]': + ) -> 'CacheManager.Result[str]': def do_get_artist_artwork( - artist_info: ArtistInfo2 - ) -> 'CacheManager.Result[Optional[str]]': + artist_info: ArtistInfo2) -> 'CacheManager.Result[str]': lastfm_url = ''.join(artist_info.largeImageUrl or []) # If it is the placeholder LastFM image, try and use the cover @@ -746,8 +750,7 @@ class CacheManager(metaclass=Singleton): ) def download_fn( - artist_info: CacheManager.Result[ArtistInfo2] - ) -> Optional[str]: + artist_info: CacheManager.Result[ArtistInfo2]) -> str: # In this case, artist_info is a future, so we have to wait for # its result before calculating. Then, immediately unwrap the # result() because we are already within a future. @@ -851,11 +854,12 @@ class CacheManager(metaclass=Singleton): def do_delete_cached_songs(): # Do the actual download. for f in map(CacheManager.get_song_details, song_ids): - relative_path = f.result().path + song: Child = f.result() + relative_path = song.path abs_path = self.calculate_abs_path(relative_path) if abs_path.exists(): abs_path.unlink() - on_song_delete() + on_song_delete(song.id) return CacheManager.create_future(do_delete_cached_songs) @@ -902,12 +906,12 @@ class CacheManager(metaclass=Singleton): return CacheManager.create_future(do_batch_download_songs) def get_cover_art_filename( - self, - id: str, - before_download: Callable[[], None] = lambda: None, - force: bool = False, - allow_download: bool = True, - ) -> 'CacheManager.Result[Optional[str]]': + self, + id: str, + before_download: Callable[[], None] = lambda: None, + force: bool = False, + allow_download: bool = True, + ) -> 'CacheManager.Result[str]': if id is None: default_art_path = 'ui/images/default-album-art.png' return CacheManager.Result.from_data( diff --git a/sublime/dbus_manager.py b/sublime/dbus_manager.py index fc45186..30878ed 100644 --- a/sublime/dbus_manager.py +++ b/sublime/dbus_manager.py @@ -3,7 +3,7 @@ import os import re from collections import defaultdict -from typing import Any, Callable, Dict, Tuple +from typing import Any, Callable, DefaultDict, Dict, List, Tuple from deepdiff import DeepDiff from gi.repository import Gio, GLib @@ -17,7 +17,7 @@ def dbus_propagate(param_self: Any = None) -> Callable: """ Wraps a function which causes changes to DBus properties. """ - def decorator(function): + def decorator(function: Callable) -> Callable: @functools.wraps(function) def wrapper(*args): function(*args) @@ -36,8 +36,17 @@ class DBusManager: def __init__( self, connection: Gio.DBusConnection, - do_on_method_call, - on_set_property, + do_on_method_call: Callable[[ + Gio.DBusConnection, + str, + str, + str, + str, + GLib.Variant, + Gio.DBusMethodInvocation, + ], None], + on_set_property: Callable[ + [Gio.DBusConnection, str, str, str, str, GLib.Variant], None], get_state_and_player: Callable[[], Tuple[ApplicationState, Player]], ): self.get_state_and_player = get_state_and_player @@ -45,7 +54,7 @@ class DBusManager: self.on_set_property = on_set_property self.connection = connection - def dbus_name_acquired(connection, name): + def dbus_name_acquired(connection: Gio.DBusConnection, name: str): specs = [ 'org.mpris.MediaPlayer2.xml', 'org.mpris.MediaPlayer2.Player.xml', @@ -84,12 +93,12 @@ class DBusManager: Gio.bus_unown_name(self.bus_number) def on_get_property( - self, - connection: Gio.DBusConnection, - sender, - path, - interface: str, - property_name: str, + self, + connection: Gio.DBusConnection, + sender: str, + path: str, + interface: str, + property_name: str, ) -> GLib.Variant: value = self.property_dict().get(interface, {}).get(property_name) return DBusManager.to_variant(value) @@ -97,12 +106,12 @@ class DBusManager: def on_method_call( self, connection: Gio.DBusConnection, - sender, - path, + sender: str, + path: str, interface: str, method: str, - params, - invocation, + params: GLib.Variant, + invocation: Gio.DBusMethodInvocation, ): # TODO (#127): I don't even know if this works. if interface == 'org.freedesktop.DBus.Properties': @@ -267,7 +276,11 @@ class DBusManager: }, } - def get_mpris_metadata(self, idx: int, play_queue): + def get_mpris_metadata( + self, + idx: int, + play_queue: List[str], + ) -> Dict[str, Any]: song_result = CacheManager.get_song_details(play_queue[idx]) if song_result.is_future: return {} @@ -289,8 +302,8 @@ class DBusManager: 'xesam:title': song.title, } - def get_dbus_playlist(self, play_queue): - seen_counts = defaultdict(int) + def get_dbus_playlist(self, play_queue: List[str]) -> List[str]: + seen_counts: DefaultDict[str, int] = defaultdict(int) tracks = [] for song_id in play_queue: id_ = seen_counts[song_id] diff --git a/sublime/server/server.py b/sublime/server/server.py index ef6bcd1..faf41b2 100644 --- a/sublime/server/server.py +++ b/sublime/server/server.py @@ -840,7 +840,7 @@ class Server: """ return self.do_download(self._make_url('download'), id=id) - def get_cover_art(self, id: str, size: int = 1000): + def get_cover_art(self, id: str, size: int = 1000) -> bytes: """ Returns the cover art image in binary form. @@ -850,7 +850,7 @@ class Server: return self.do_download( self._make_url('getCoverArt'), id=id, size=size) - def get_cover_art_url(self, id: str, size: int = 1000): + def get_cover_art_url(self, id: str, size: int = 1000) -> str: """ Returns the URL of the cover art image. diff --git a/sublime/ui/common/album_with_songs.py b/sublime/ui/common/album_with_songs.py index 27101a5..1f46de1 100644 --- a/sublime/ui/common/album_with_songs.py +++ b/sublime/ui/common/album_with_songs.py @@ -185,7 +185,7 @@ class AlbumWithSongs(Gtk.Box): store, paths = tree.get_selection().get_selected_rows() allow_deselect = False - def on_download_state_change(song_id: Any = None): + def on_download_state_change(song_id: int): self.update_album_songs(self.album.id) # Use the new selection instead of the old one for calculating what diff --git a/sublime/ui/player_controls.py b/sublime/ui/player_controls.py index b042b34..e580ac7 100644 --- a/sublime/ui/player_controls.py +++ b/sublime/ui/player_controls.py @@ -379,7 +379,11 @@ class PlayerControls(Gtk.ActionBar): def on_device_refresh_click(self, _: Any): self.update_device_list(force=True) - def on_play_queue_button_press(self, tree: Any, event: Gdk.EventButton): + def on_play_queue_button_press( + self, + tree: Any, + event: Gdk.EventButton, + ) -> bool: if event.button == 3: # Right click clicked_path = tree.get_path_at_pos(event.x, event.y) diff --git a/sublime/ui/playlists.py b/sublime/ui/playlists.py index 0e8d999..d9d78f0 100644 --- a/sublime/ui/playlists.py +++ b/sublime/ui/playlists.py @@ -5,7 +5,7 @@ from typing import Any, Iterable, List, Tuple import gi gi.require_version('Gtk', '3.0') from fuzzywuzzy import process -from gi.repository import Gio, GLib, GObject, Gtk, Pango +from gi.repository import Gdk, Gio, GLib, GObject, Gtk, Pango from sublime.cache_manager import CacheManager from sublime.server.api_objects import PlaylistWithSongs @@ -362,7 +362,13 @@ class PlaylistDetailPanel(Gtk.Overlay): def max_score_for_key(key: str, rows: Tuple) -> int: return max(row_score(key, row) for row in rows) - def playlist_song_list_search_fn(model, col, key, treeiter, data=None): + def playlist_song_list_search_fn( + model: Gtk.ListStore, + col: int, + key: str, + treeiter: Gtk.TreeIter, + data: Any = None, + ) -> bool: # TODO (#28): this is very inefficient, it's slow when the result # is close to the bottom of the list. Would be good to research # what the default one does (maybe it uses an index?). @@ -608,7 +614,7 @@ class PlaylistDetailPanel(Gtk.Overlay): }, ) - def on_song_activated(self, treeview, idx, column): + def on_song_activated(self, _: Any, idx: Gtk.TreePath, col: Any): # The song ID is in the last column of the model. self.emit( 'song-clicked', @@ -619,7 +625,11 @@ class PlaylistDetailPanel(Gtk.Overlay): }, ) - def on_song_button_press(self, tree, event): + def on_song_button_press( + self, + tree: Gtk.TreeView, + event: Gdk.EventButton, + ) -> bool: if event.button == 3: # Right click clicked_path = tree.get_path_at_pos(event.x, event.y) if not clicked_path: @@ -628,7 +638,7 @@ class PlaylistDetailPanel(Gtk.Overlay): store, paths = tree.get_selection().get_selected_rows() allow_deselect = False - def on_download_state_change(**kwargs): + def on_download_state_change(song_id: int): GLib.idle_add( lambda: self.update_playlist_view( self.playlist_id, @@ -678,6 +688,8 @@ class PlaylistDetailPanel(Gtk.Overlay): if not allow_deselect: return True + return False + def on_playlist_model_row_move(self, *args): # If we are programatically editing the song list, don't do anything. if self.editing_playlist_song_list: diff --git a/sublime/ui/util.py b/sublime/ui/util.py index 9102582..1141b06 100644 --- a/sublime/ui/util.py +++ b/sublime/ui/util.py @@ -150,7 +150,7 @@ def diff_model_store(store_to_edit: Any, new_store: Iterable[Any]): def show_song_popover( - song_ids, + song_ids: List[int], x: int, y: int, relative_to: Any, @@ -309,7 +309,7 @@ def async_callback( future_fn: Callable[..., Future], before_download: Callable[[Any], None] = None, on_failure: Callable[[Any, Exception], None] = None, -): +) -> Callable[[Callable], Callable]: """ Defines the ``async_callback`` decorator. @@ -321,10 +321,10 @@ def async_callback( :param future_fn: a function which generates a :class:`concurrent.futures.Future` or :class:`CacheManager.Result`. """ - def decorator(callback_fn): + def decorator(callback_fn: Callable) -> Callable: @functools.wraps(callback_fn) def wrapper( - self, + self: Any, *args, state: ApplicationState = None, force: bool = False,