Coverage for sublime_music/app.py : 11%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import logging
2import os
3import random
4import shutil
5import sys
6from datetime import timedelta
7from functools import partial
8from pathlib import Path
9from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Tuple
10from urllib.parse import urlparse
12import bleach
14try:
15 import osxmmkeys
17 tap_imported = True
18except Exception:
19 tap_imported = False
21from gi.repository import Gdk, GdkPixbuf, Gio, GLib, Gtk
23try:
24 import gi
26 gi.require_version("Notify", "0.7")
27 from gi.repository import Notify
29 glib_notify_exists = True
30except Exception:
31 # I really don't care what kind of exception it is, all that matters is the
32 # import failed for some reason.
33 logging.warning(
34 "Unable to import Notify from GLib. Notifications will be disabled."
35 )
36 glib_notify_exists = False
38from .adapters import (
39 AdapterManager,
40 AlbumSearchQuery,
41 CacheMissError,
42 DownloadProgress,
43 Result,
44 SongCacheStatus,
45)
46from .adapters.api_objects import Playlist, PlayQueue, Song
47from .config import AppConfiguration, ProviderConfiguration
48from .dbus import dbus_propagate, DBusManager
49from .players import PlayerDeviceEvent, PlayerEvent, PlayerManager
50from .ui.configure_provider import ConfigureProviderDialog
51from .ui.main import MainWindow
52from .ui.state import RepeatType, UIState
53from .util import resolve_path
56class SublimeMusicApp(Gtk.Application):
57 def __init__(self, config_file: Path):
58 super().__init__(application_id="app.sublimemusic.SublimeMusic")
59 if glib_notify_exists:
60 Notify.init("Sublime Music")
62 self.window: Optional[Gtk.Window] = None
63 self.app_config = AppConfiguration.load_from_file(config_file)
64 self.dbus_manager: Optional[DBusManager] = None
66 self.connect("shutdown", self.on_app_shutdown)
68 player_manager: Optional[PlayerManager] = None
69 exiting: bool = False
71 def do_startup(self):
72 Gtk.Application.do_startup(self)
74 def add_action(name: str, fn: Callable, parameter_type: str = None):
75 """Registers an action with the application."""
76 if type(parameter_type) == str:
77 parameter_type = GLib.VariantType(parameter_type)
78 action = Gio.SimpleAction.new(name, parameter_type)
79 action.connect("activate", fn)
80 self.add_action(action)
82 # Add action for menu items.
83 add_action("add-new-music-provider", self.on_add_new_music_provider)
84 add_action("edit-current-music-provider", self.on_edit_current_music_provider)
85 add_action(
86 "switch-music-provider", self.on_switch_music_provider, parameter_type="s"
87 )
88 add_action(
89 "remove-music-provider", self.on_remove_music_provider, parameter_type="s"
90 )
92 # Add actions for player controls
93 add_action("play-pause", self.on_play_pause)
94 add_action("next-track", self.on_next_track)
95 add_action("prev-track", self.on_prev_track)
96 add_action("repeat-press", self.on_repeat_press)
97 add_action("shuffle-press", self.on_shuffle_press)
99 # Navigation actions.
100 add_action("play-next", self.on_play_next, parameter_type="as")
101 add_action("add-to-queue", self.on_add_to_queue, parameter_type="as")
102 add_action("go-to-album", self.on_go_to_album, parameter_type="s")
103 add_action("go-to-artist", self.on_go_to_artist, parameter_type="s")
104 add_action("browse-to", self.browse_to, parameter_type="s")
105 add_action("go-to-playlist", self.on_go_to_playlist, parameter_type="s")
107 add_action("go-online", self.on_go_online)
108 add_action("refresh-devices", self.on_refresh_devices)
109 add_action(
110 "refresh-window",
111 lambda *a: self.on_refresh_window(None, {}, True),
112 )
113 add_action("mute-toggle", self.on_mute_toggle)
114 add_action(
115 "update-play-queue-from-server",
116 lambda a, p: self.update_play_state_from_server(),
117 )
119 if tap_imported:
120 self.tap = osxmmkeys.Tap()
121 self.tap.on("play_pause", self.on_play_pause)
122 self.tap.on("next_track", self.on_next_track)
123 self.tap.on("prev_track", self.on_prev_track)
124 self.tap.start()
126 def do_activate(self):
127 # We only allow a single window and raise any existing ones
128 if self.window:
129 self.window.present()
130 return
132 # Configure Icons
133 default_icon_theme = Gtk.IconTheme.get_default()
134 for adapter in AdapterManager.available_adapters:
135 if icon_dir := adapter.get_ui_info().icon_dir:
136 default_icon_theme.append_search_path(str(icon_dir))
138 icon_dirs = [resolve_path("ui/icons"), resolve_path("adapters/icons")]
139 for icon_dir in icon_dirs:
140 default_icon_theme.append_search_path(str(icon_dir))
142 # Windows are associated with the application when the last one is
143 # closed the application shuts down.
144 self.window = MainWindow(application=self, title="Sublime Music")
146 # Configure the CSS provider so that we can style elements on the
147 # window.
148 css_provider = Gtk.CssProvider()
149 css_provider.load_from_path(str(resolve_path("ui/app_styles.css")))
150 context = Gtk.StyleContext()
151 screen = Gdk.Screen.get_default()
152 context.add_provider_for_screen(
153 screen, css_provider, Gtk.STYLE_PROVIDER_PRIORITY_USER
154 )
156 self.window.show_all()
157 self.window.present()
159 # Load the state for the server, if it exists.
160 self.app_config.load_state()
162 # If there is no current provider, use the first one if there are any
163 # configured, and if none are configured, then show the dialog to create a new
164 # one.
165 if self.app_config.provider is None:
166 if len(self.app_config.providers) == 0:
167 self.show_configure_servers_dialog()
169 # If they didn't add one with the dialog, close the window.
170 if len(self.app_config.providers) == 0:
171 self.window.close()
172 return
174 AdapterManager.reset(self.app_config, self.on_song_download_progress)
176 # Connect after we know there's a server configured.
177 self.window.stack.connect("notify::visible-child", self.on_stack_change)
178 self.window.connect("song-clicked", self.on_song_clicked)
179 self.window.connect("songs-removed", self.on_songs_removed)
180 self.window.connect("refresh-window", self.on_refresh_window)
181 self.window.connect("notification-closed", self.on_notification_closed)
182 self.window.connect("go-to", self.on_window_go_to)
183 self.window.connect("key-press-event", self.on_window_key_press)
184 self.window.player_controls.connect("song-scrub", self.on_song_scrub)
185 self.window.player_controls.connect("device-update", self.on_device_update)
186 self.window.player_controls.connect("volume-change", self.on_volume_change)
188 # Configure the players
189 self.last_play_queue_update = timedelta(0)
190 self.loading_state = False
191 self.should_scrobble_song = False
193 def on_timepos_change(value: Optional[float]):
194 if (
195 self.loading_state
196 or not self.window
197 or not self.app_config.state.current_song
198 ):
199 return
201 if value is None:
202 self.last_play_queue_update = timedelta(0)
203 return
205 self.app_config.state.song_progress = timedelta(seconds=value)
206 GLib.idle_add(
207 self.window.player_controls.update_scrubber,
208 self.app_config.state.song_progress,
209 self.app_config.state.current_song.duration,
210 self.app_config.state.song_stream_cache_progress,
211 )
213 if (self.last_play_queue_update + timedelta(15)).total_seconds() <= value:
214 self.save_play_queue()
216 if (
217 value > 5
218 and self.should_scrobble_song
219 and AdapterManager.can_scrobble_song()
220 ):
221 AdapterManager.scrobble_song(self.app_config.state.current_song)
222 self.should_scrobble_song = False
224 def on_track_end():
225 at_end = self.app_config.state.next_song_index is None
226 no_repeat = self.app_config.state.repeat_type == RepeatType.NO_REPEAT
227 if at_end and no_repeat:
228 self.app_config.state.playing = False
229 self.app_config.state.current_song_index = -1
230 self.update_window()
231 return
233 GLib.idle_add(self.on_next_track)
235 def on_player_event(event: PlayerEvent):
236 if event.type == PlayerEvent.EventType.PLAY_STATE_CHANGE:
237 assert event.playing is not None
238 self.app_config.state.playing = event.playing
239 if self.dbus_manager:
240 self.dbus_manager.property_diff()
241 self.update_window()
243 elif event.type == PlayerEvent.EventType.VOLUME_CHANGE:
244 assert event.volume is not None
245 self.app_config.state.volume = event.volume
246 if self.dbus_manager:
247 self.dbus_manager.property_diff()
248 self.update_window()
250 elif event.type == PlayerEvent.EventType.STREAM_CACHE_PROGRESS_CHANGE:
251 if (
252 self.loading_state
253 or not self.window
254 or not self.app_config.state.current_song
255 or event.stream_cache_duration is None
256 ):
257 return
258 self.app_config.state.song_stream_cache_progress = timedelta(
259 seconds=event.stream_cache_duration
260 )
261 GLib.idle_add(
262 self.window.player_controls.update_scrubber,
263 self.app_config.state.song_progress,
264 self.app_config.state.current_song.duration,
265 self.app_config.state.song_stream_cache_progress,
266 )
268 elif event.type == PlayerEvent.EventType.DISCONNECT:
269 assert self.player_manager
270 self.app_config.state.current_device = "this device"
271 self.player_manager.set_current_device_id(
272 self.app_config.state.current_device
273 )
274 self.player_manager.set_volume(self.app_config.state.volume)
275 self.update_window()
277 def player_device_change_callback(event: PlayerDeviceEvent):
278 assert self.player_manager
279 state_device = self.app_config.state.current_device
281 if event.delta == PlayerDeviceEvent.Delta.ADD:
282 # If the device added is the one that's supposed to be active, activate
283 # it and set the volume.
284 if event.id == state_device:
285 self.player_manager.set_current_device_id(
286 self.app_config.state.current_device
287 )
288 self.player_manager.set_volume(self.app_config.state.volume)
289 self.app_config.state.connected_device_name = event.name
290 self.app_config.state.connecting_to_device = False
291 self.app_config.state.available_players[event.player_type].add(
292 (event.id, event.name)
293 )
295 elif event.delta == PlayerDeviceEvent.Delta.REMOVE:
296 if state_device == event.id:
297 self.player_manager.pause()
298 self.app_config.state.available_players[event.player_type].remove(
299 (event.id, event.name)
300 )
302 self.update_window()
304 self.app_config.state.connecting_to_device = True
306 def check_if_connected():
307 if not self.app_config.state.connecting_to_device:
308 return
309 self.app_config.state.current_device = "this device"
310 self.app_config.state.connecting_to_device = False
311 self.player_manager.set_current_device_id(
312 self.app_config.state.current_device
313 )
314 self.update_window()
316 self.player_manager = PlayerManager(
317 on_timepos_change,
318 on_track_end,
319 on_player_event,
320 lambda *a: GLib.idle_add(player_device_change_callback, *a),
321 self.app_config.player_config,
322 )
323 GLib.timeout_add(10000, check_if_connected)
325 # Update after Adapter Initial Sync
326 def after_initial_sync(_):
327 self.update_window()
329 # Prompt to load the play queue from the server.
330 if AdapterManager.can_get_play_queue():
331 self.update_play_state_from_server(prompt_confirm=True)
333 # Get the playlists, just so that we don't have tons of cache misses from
334 # DBus trying to get the playlists.
335 if AdapterManager.can_get_playlists():
336 AdapterManager.get_playlists()
338 inital_sync_result = AdapterManager.initial_sync()
339 inital_sync_result.add_done_callback(after_initial_sync)
341 # Send out to the bus that we exist.
342 if self.dbus_manager:
343 self.dbus_manager.property_diff()
345 # ########## DBUS MANAGMENT ########## #
346 def do_dbus_register(self, connection: Gio.DBusConnection, path: str) -> bool:
347 self.dbus_manager = DBusManager(
348 connection,
349 self.on_dbus_method_call,
350 self.on_dbus_set_property,
351 lambda: (self.app_config, self.player_manager),
352 )
353 return True
355 def on_dbus_method_call(
356 self,
357 connection: Gio.DBusConnection,
358 sender: str,
359 path: str,
360 interface: str,
361 method: str,
362 params: GLib.Variant,
363 invocation: Gio.DBusMethodInvocation,
364 ):
365 def seek_fn(offset: float):
366 if not self.app_config.state.current_song:
367 return
368 new_seconds = self.app_config.state.song_progress + timedelta(
369 microseconds=offset
370 )
372 # This should not ever happen. The current_song should always have
373 # a duration, but the Child object has `duration` optional because
374 # it could be a directory.
375 assert self.app_config.state.current_song.duration is not None
376 self.on_song_scrub(
377 None,
378 (
379 new_seconds.total_seconds()
380 / self.app_config.state.current_song.duration.total_seconds()
381 )
382 * 100,
383 )
385 def set_pos_fn(track_id: str, position: float = 0):
386 if self.app_config.state.playing:
387 self.on_play_pause()
388 pos_seconds = timedelta(microseconds=position)
389 self.app_config.state.song_progress = pos_seconds
390 track_id, occurrence = track_id.split("/")[-2:]
392 # Find the (N-1)th time that the track id shows up in the list. (N
393 # is the -*** suffix on the track id.)
394 song_index = [
395 i
396 for i, x in enumerate(self.app_config.state.play_queue)
397 if x == track_id
398 ][int(occurrence) or 0]
400 self.play_song(song_index)
402 def get_tracks_metadata(track_ids: List[str]) -> GLib.Variant:
403 if not self.dbus_manager:
404 return
406 if len(track_ids) == 0:
407 # We are lucky, just return an empty list.
408 return GLib.Variant("(aa{sv})", ([],))
410 # Have to calculate all of the metadatas so that we can deal with
411 # repeat song IDs.
412 metadatas: Iterable[Any] = [
413 self.dbus_manager.get_mpris_metadata(
414 i,
415 self.app_config.state.play_queue,
416 )
417 for i in range(len(self.app_config.state.play_queue))
418 ]
420 # Get rid of all of the tracks that were not requested.
421 metadatas = list(
422 filter(lambda m: m["mpris:trackid"] in track_ids, metadatas)
423 )
425 assert len(metadatas) == len(track_ids)
427 # Sort them so they get returned in the same order as they were
428 # requested.
429 metadatas = sorted(
430 metadatas, key=lambda m: track_ids.index(m["mpris:trackid"])
431 )
433 # Turn them into dictionaries that can actually be serialized into
434 # a GLib.Variant.
435 metadatas = map(
436 lambda m: {k: DBusManager.to_variant(v) for k, v in m.items()},
437 metadatas,
438 )
440 return GLib.Variant("(aa{sv})", (list(metadatas),))
442 def activate_playlist(playlist_id: str):
443 playlist_id = playlist_id.split("/")[-1]
444 playlist = AdapterManager.get_playlist_details(playlist_id).result()
446 # Calculate the song id to play.
447 song_idx = 0
448 if self.app_config.state.shuffle_on:
449 song_idx = random.randint(0, len(playlist.songs) - 1)
451 self.on_song_clicked(
452 None,
453 song_idx,
454 tuple(s.id for s in playlist.songs),
455 {"active_playlist_id": playlist_id},
456 )
458 def get_playlists(
459 index: int,
460 max_count: int,
461 order: str,
462 reverse_order: bool,
463 ) -> GLib.Variant:
464 playlists_result = AdapterManager.get_playlists()
465 if not playlists_result.data_is_available:
466 # We don't want to wait for the response in this case, so just
467 # return an empty array.
468 return GLib.Variant("(a(oss))", ([],))
470 playlists = list(playlists_result.result())
472 sorters = {
473 "Alphabetical": lambda p: p.name,
474 "Created": lambda p: p.created,
475 "Modified": lambda p: p.changed,
476 }
477 if order in sorters:
478 playlists.sort(
479 key=sorters.get(order, lambda p: p),
480 reverse=reverse_order,
481 )
483 def make_playlist_tuple(p: Playlist) -> GLib.Variant:
484 cover_art_filename = AdapterManager.get_cover_art_uri(
485 p.cover_art,
486 "file",
487 allow_download=False,
488 ).result()
489 return (f"/playlist/{p.id}", p.name, cover_art_filename or "")
491 return GLib.Variant(
492 "(a(oss))",
493 (
494 [
495 make_playlist_tuple(p)
496 for p in playlists[index : (index + max_count)]
497 ],
498 ),
499 )
501 def play():
502 if not self.app_config.state.playing:
503 self.on_play_pause()
505 def pause():
506 if self.app_config.state.playing:
507 self.on_play_pause()
509 method_call_map: Dict[str, Dict[str, Any]] = {
510 "org.mpris.MediaPlayer2": {
511 "Raise": self.window and self.window.present,
512 "Quit": self.window and self.window.destroy,
513 },
514 "org.mpris.MediaPlayer2.Player": {
515 "Next": self.on_next_track,
516 "Previous": self.on_prev_track,
517 "Pause": pause,
518 "PlayPause": self.on_play_pause,
519 "Stop": pause,
520 "Play": play,
521 "Seek": seek_fn,
522 "SetPosition": set_pos_fn,
523 },
524 "org.mpris.MediaPlayer2.TrackList": {
525 "GoTo": set_pos_fn,
526 "GetTracksMetadata": get_tracks_metadata,
527 # 'RemoveTrack': remove_track,
528 },
529 "org.mpris.MediaPlayer2.Playlists": {
530 "ActivatePlaylist": activate_playlist,
531 "GetPlaylists": get_playlists,
532 },
533 }
534 method_fn = method_call_map.get(interface, {}).get(method)
535 if method_fn is None:
536 logging.warning(f"Unknown/unimplemented method: {interface}.{method}.")
537 invocation.return_value(method_fn(*params) if callable(method_fn) else None)
539 def on_dbus_set_property(
540 self,
541 connection: Gio.DBusConnection,
542 sender: str,
543 path: str,
544 interface: str,
545 property_name: str,
546 value: GLib.Variant,
547 ):
548 def change_loop(new_loop_status: GLib.Variant):
549 self.app_config.state.repeat_type = RepeatType.from_mpris_loop_status(
550 new_loop_status.get_string()
551 )
552 self.update_window()
554 def set_shuffle(new_val: GLib.Variant):
555 if new_val.get_boolean() != self.app_config.state.shuffle_on:
556 self.on_shuffle_press(None, None)
558 def set_volume(new_val: GLib.Variant):
559 self.on_volume_change(None, new_val.get_double() * 100)
561 setter_map: Dict[str, Dict[str, Any]] = {
562 "org.mpris.MediaPlayer2.Player": {
563 "LoopStatus": change_loop,
564 "Rate": lambda _: None,
565 "Shuffle": set_shuffle,
566 "Volume": set_volume,
567 }
568 }
570 setter = setter_map.get(interface, {}).get(property_name)
571 if setter is None:
572 logging.warning("Set: Unknown property: {property_name}.")
573 return
574 if callable(setter):
575 setter(value)
577 # ########## ACTION HANDLERS ########## #
578 @dbus_propagate()
579 def on_refresh_window(self, _, state_updates: Dict[str, Any], force: bool = False):
580 if settings := state_updates.get("__settings__"):
581 for k, v in settings.items():
582 setattr(self.app_config, k, v)
583 if (offline_mode := settings.get("offline_mode")) is not None:
584 AdapterManager.on_offline_mode_change(offline_mode)
586 del state_updates["__settings__"]
587 self.app_config.save()
589 if player_setting := state_updates.get("__player_setting__"):
590 player_name, option_name, value = player_setting
591 self.app_config.player_config[player_name][option_name] = value
592 del state_updates["__player_setting__"]
593 if pm := self.player_manager:
594 pm.change_settings(self.app_config.player_config)
595 self.app_config.save()
597 for k, v in state_updates.items():
598 setattr(self.app_config.state, k, v)
599 self.update_window(force=force)
601 def on_notification_closed(self, _):
602 self.app_config.state.current_notification = None
603 self.update_window()
605 def on_add_new_music_provider(self, *args):
606 self.show_configure_servers_dialog()
608 def on_edit_current_music_provider(self, *args):
609 self.show_configure_servers_dialog(self.app_config.provider.clone())
611 def on_switch_music_provider(self, _, provider_id: GLib.Variant):
612 if self.app_config.state.playing:
613 self.on_play_pause()
614 self.app_config.save()
615 self.app_config.current_provider_id = provider_id.get_string()
616 self.reset_state()
617 self.app_config.save()
619 def on_remove_music_provider(self, _, provider_id: GLib.Variant):
620 provider = self.app_config.providers[provider_id.get_string()]
621 confirm_dialog = Gtk.MessageDialog(
622 transient_for=self.window,
623 message_type=Gtk.MessageType.WARNING,
624 buttons=(
625 Gtk.STOCK_CANCEL,
626 Gtk.ResponseType.CANCEL,
627 Gtk.STOCK_DELETE,
628 Gtk.ResponseType.YES,
629 ),
630 text=f"Are you sure you want to delete the {provider.name} music provider?",
631 )
632 confirm_dialog.format_secondary_markup(
633 "Deleting this music provider will delete all cached songs and metadata "
634 "associated with this provider."
635 )
636 if confirm_dialog.run() == Gtk.ResponseType.YES:
637 assert self.app_config.cache_location
638 provider_dir = self.app_config.cache_location.joinpath(provider.id)
639 shutil.rmtree(str(provider_dir), ignore_errors=True)
640 del self.app_config.providers[provider.id]
642 confirm_dialog.destroy()
644 def on_window_go_to(self, win: Any, action: str, value: str):
645 {
646 "album": self.on_go_to_album,
647 "artist": self.on_go_to_artist,
648 "playlist": self.on_go_to_playlist,
649 }[action](None, GLib.Variant("s", value))
651 @dbus_propagate()
652 def on_play_pause(self, *args):
653 if self.app_config.state.current_song_index < 0:
654 return
656 self.app_config.state.playing = not self.app_config.state.playing
658 if self.player_manager.song_loaded and (
659 self.player_manager.current_song == self.app_config.state.current_song
660 ):
661 self.player_manager.toggle_play()
662 self.save_play_queue()
663 else:
664 # This is from a restart, start playing the file.
665 self.play_song(self.app_config.state.current_song_index)
667 self.update_window()
669 def on_next_track(self, *args):
670 if self.app_config.state.current_song is None:
671 # This may happen due to DBUS, ignore.
672 return
674 song_index_to_play = self.app_config.state.next_song_index
675 if song_index_to_play is None:
676 # We may end up here due to D-Bus.
677 return
679 self.app_config.state.current_song_index = song_index_to_play
680 self.app_config.state.song_progress = timedelta(0)
681 if self.app_config.state.playing:
682 self.play_song(song_index_to_play, reset=True)
683 else:
684 self.update_window()
686 def on_prev_track(self, *args):
687 if self.app_config.state.current_song is None:
688 # This may happen due to DBUS, ignore.
689 return
690 # Go back to the beginning of the song if we are past 5 seconds.
691 # Otherwise, go to the previous song.
692 no_repeat = self.app_config.state.repeat_type == RepeatType.NO_REPEAT
694 if self.app_config.state.repeat_type == RepeatType.REPEAT_SONG:
695 song_index_to_play = self.app_config.state.current_song_index
696 elif self.app_config.state.song_progress.total_seconds() < 5:
697 if self.app_config.state.current_song_index == 0 and no_repeat:
698 song_index_to_play = 0
699 else:
700 song_index_to_play = (
701 self.app_config.state.current_song_index - 1
702 ) % len(self.app_config.state.play_queue)
703 else:
704 # Go back to the beginning of the song.
705 song_index_to_play = self.app_config.state.current_song_index
707 self.app_config.state.current_song_index = song_index_to_play
708 self.app_config.state.song_progress = timedelta(0)
709 if self.player_manager.playing:
710 self.play_song(
711 song_index_to_play,
712 reset=True,
713 # search backwards for a song to play if offline
714 playable_song_search_direction=-1,
715 )
716 else:
717 self.update_window()
719 @dbus_propagate()
720 def on_repeat_press(self, *args):
721 # Cycle through the repeat types.
722 new_repeat_type = RepeatType((self.app_config.state.repeat_type.value + 1) % 3)
723 self.app_config.state.repeat_type = new_repeat_type
724 self.update_window()
726 @dbus_propagate()
727 def on_shuffle_press(self, *args):
728 if self.app_config.state.shuffle_on:
729 # Revert to the old play queue.
730 old_play_queue_copy = self.app_config.state.old_play_queue
731 self.app_config.state.current_song_index = old_play_queue_copy.index(
732 self.app_config.state.current_song.id
733 )
734 self.app_config.state.play_queue = old_play_queue_copy
735 else:
736 self.app_config.state.old_play_queue = self.app_config.state.play_queue
738 mutable_play_queue = list(self.app_config.state.play_queue)
740 # Remove the current song, then shuffle and put the song back.
741 song_id = self.app_config.state.current_song.id
742 del mutable_play_queue[self.app_config.state.current_song_index]
743 random.shuffle(mutable_play_queue)
744 self.app_config.state.play_queue = (song_id,) + tuple(mutable_play_queue)
745 self.app_config.state.current_song_index = 0
747 self.app_config.state.shuffle_on = not self.app_config.state.shuffle_on
748 self.update_window()
750 @dbus_propagate()
751 def on_play_next(self, action: Any, song_ids: GLib.Variant):
752 song_ids = tuple(song_ids)
753 if self.app_config.state.current_song is None:
754 insert_at = 0
755 else:
756 insert_at = self.app_config.state.current_song_index + 1
758 self.app_config.state.play_queue = (
759 self.app_config.state.play_queue[:insert_at]
760 + song_ids
761 + self.app_config.state.play_queue[insert_at:]
762 )
763 self.app_config.state.old_play_queue += song_ids
764 self.update_window()
766 @dbus_propagate()
767 def on_add_to_queue(self, action: Any, song_ids: GLib.Variant):
768 song_ids = tuple(song_ids)
769 self.app_config.state.play_queue += tuple(song_ids)
770 self.app_config.state.old_play_queue += tuple(song_ids)
771 self.update_window()
773 def on_go_to_album(self, action: Any, album_id: GLib.Variant):
774 # Switch to the Alphabetical by Name view to guarantee that the album is there.
775 self.app_config.state.current_album_search_query = AlbumSearchQuery(
776 AlbumSearchQuery.Type.ALPHABETICAL_BY_NAME,
777 genre=self.app_config.state.current_album_search_query.genre,
778 year_range=self.app_config.state.current_album_search_query.year_range,
779 )
781 self.app_config.state.current_tab = "albums"
782 self.app_config.state.selected_album_id = album_id.get_string()
783 self.update_window()
785 def on_go_to_artist(self, action: Any, artist_id: GLib.Variant):
786 self.app_config.state.current_tab = "artists"
787 self.app_config.state.selected_artist_id = artist_id.get_string()
788 self.update_window()
790 def browse_to(self, action: Any, item_id: GLib.Variant):
791 self.app_config.state.current_tab = "browse"
792 self.app_config.state.selected_browse_element_id = item_id.get_string()
793 self.update_window()
795 def on_go_to_playlist(self, action: Any, playlist_id: GLib.Variant):
796 self.app_config.state.current_tab = "playlists"
797 self.app_config.state.selected_playlist_id = playlist_id.get_string()
798 self.update_window()
800 def on_go_online(self, *args):
801 self.on_refresh_window(None, {"__settings__": {"offline_mode": False}})
803 def on_refresh_devices(self, *args):
804 self.player_manager.refresh_players()
806 def reset_state(self):
807 if self.app_config.state.playing:
808 self.on_play_pause()
809 self.loading_state = True
810 self.player_manager.reset()
811 AdapterManager.reset(self.app_config, self.on_song_download_progress)
812 self.loading_state = False
814 # Update the window according to the new server configuration.
815 self.update_window()
817 def on_stack_change(self, stack: Gtk.Stack, _):
818 self.app_config.state.current_tab = stack.get_visible_child_name()
819 self.update_window()
821 def on_song_clicked(
822 self,
823 win: Any,
824 song_index: int,
825 song_queue: Tuple[str, ...],
826 metadata: Dict[str, Any],
827 ):
828 song_queue = tuple(song_queue)
829 # Reset the play queue so that we don't ever revert back to the
830 # previous one.
831 old_play_queue = song_queue
833 if (force_shuffle := metadata.get("force_shuffle_state")) is not None:
834 self.app_config.state.shuffle_on = force_shuffle
836 self.app_config.state.active_playlist_id = metadata.get("active_playlist_id")
838 # If shuffle is enabled, then shuffle the playlist.
839 if self.app_config.state.shuffle_on and not metadata.get("no_reshuffle"):
840 song_id = song_queue[song_index]
841 song_queue_list = list(
842 song_queue[:song_index] + song_queue[song_index + 1 :]
843 )
844 random.shuffle(song_queue_list)
845 song_queue = (song_id, *song_queue_list)
846 song_index = 0
848 self.play_song(
849 song_index,
850 reset=True,
851 old_play_queue=old_play_queue,
852 play_queue=song_queue,
853 )
855 def on_songs_removed(self, win: Any, song_indexes_to_remove: List[int]):
856 self.app_config.state.play_queue = tuple(
857 song_id
858 for i, song_id in enumerate(self.app_config.state.play_queue)
859 if i not in song_indexes_to_remove
860 )
862 # Determine how many songs before the currently playing one were also
863 # deleted.
864 before_current = [
865 i
866 for i in song_indexes_to_remove
867 if i < self.app_config.state.current_song_index
868 ]
870 if self.app_config.state.current_song_index in song_indexes_to_remove:
871 if len(self.app_config.state.play_queue) == 0:
872 self.on_play_pause()
873 self.app_config.state.current_song_index = -1
874 self.update_window()
875 return
877 self.app_config.state.current_song_index -= len(before_current)
878 self.play_song(self.app_config.state.current_song_index, reset=True)
879 else:
880 self.app_config.state.current_song_index -= len(before_current)
881 self.update_window()
882 self.save_play_queue()
884 @dbus_propagate()
885 def on_song_scrub(self, _, scrub_value: float):
886 if not self.app_config.state.current_song or not self.window:
887 return
889 # This should not ever happen. The current_song should always have
890 # a duration, but the Child object has `duration` optional because
891 # it could be a directory.
892 assert self.app_config.state.current_song.duration is not None
893 new_time = self.app_config.state.current_song.duration * (scrub_value / 100)
895 self.app_config.state.song_progress = new_time
896 self.window.player_controls.update_scrubber(
897 self.app_config.state.song_progress,
898 self.app_config.state.current_song.duration,
899 self.app_config.state.song_stream_cache_progress,
900 )
902 # If already playing, then make the player itself seek.
903 if self.player_manager and self.player_manager.song_loaded:
904 self.player_manager.seek(new_time)
906 self.save_play_queue()
908 def on_device_update(self, _, device_id: str):
909 assert self.player_manager
910 if device_id == self.app_config.state.current_device:
911 return
912 self.app_config.state.current_device = device_id
914 if was_playing := self.app_config.state.playing:
915 self.on_play_pause()
917 self.player_manager.set_current_device_id(self.app_config.state.current_device)
919 if self.dbus_manager:
920 self.dbus_manager.property_diff()
922 self.update_window()
924 if was_playing:
925 self.on_play_pause()
926 if self.dbus_manager:
927 self.dbus_manager.property_diff()
929 @dbus_propagate()
930 def on_mute_toggle(self, *args):
931 self.app_config.state.is_muted = not self.app_config.state.is_muted
932 self.player_manager.set_muted(self.app_config.state.is_muted)
933 self.update_window()
935 @dbus_propagate()
936 def on_volume_change(self, _, value: float):
937 assert self.player_manager
938 self.app_config.state.volume = value
939 self.player_manager.set_volume(self.app_config.state.volume)
940 self.update_window()
942 def on_window_key_press(self, window: Gtk.Window, event: Gdk.EventKey) -> bool:
943 # Need to use bitwise & here to see if CTRL is pressed.
944 if event.keyval == 102 and event.state & Gdk.ModifierType.CONTROL_MASK:
945 # Ctrl + F
946 window.search_entry.grab_focus()
947 return False
949 if event.keyval == 113 and event.state & Gdk.ModifierType.CONTROL_MASK:
950 # Ctrl + Q
951 window.destroy()
952 return False
954 # Allow spaces to work in the text entry boxes.
955 if (
956 window.search_entry.has_focus()
957 or window.playlists_panel.playlist_list.new_playlist_entry.has_focus()
958 ):
959 return False
961 # Spacebar, home/prev
962 keymap = {
963 32: self.on_play_pause,
964 65360: self.on_prev_track,
965 65367: self.on_next_track,
966 }
968 action = keymap.get(event.keyval)
969 if action:
970 action()
971 return True
973 return False
975 def on_song_download_progress(self, song_id: str, progress: DownloadProgress):
976 assert self.window
977 GLib.idle_add(self.window.update_song_download_progress, song_id, progress)
979 def on_app_shutdown(self, app: "SublimeMusicApp"):
980 self.exiting = True
981 if glib_notify_exists:
982 Notify.uninit()
984 if tap_imported and self.tap:
985 self.tap.stop()
987 if self.app_config.provider is None:
988 return
990 if self.player_manager:
991 if self.app_config.state.playing:
992 self.save_play_queue()
993 self.player_manager.pause()
994 self.player_manager.shutdown()
996 self.app_config.save()
997 if self.dbus_manager:
998 self.dbus_manager.shutdown()
999 AdapterManager.shutdown()
1001 # ########## HELPER METHODS ########## #
1002 def show_configure_servers_dialog(
1003 self,
1004 provider_config: Optional[ProviderConfiguration] = None,
1005 ):
1006 """Show the Connect to Server dialog."""
1007 dialog = ConfigureProviderDialog(self.window, provider_config)
1008 result = dialog.run()
1009 if result == Gtk.ResponseType.APPLY:
1010 assert dialog.provider_config is not None
1011 provider_id = dialog.provider_config.id
1012 dialog.provider_config.persist_secrets()
1013 self.app_config.providers[provider_id] = dialog.provider_config
1014 self.app_config.save()
1016 if provider_id == self.app_config.current_provider_id:
1017 # Just update the window.
1018 self.update_window()
1019 else:
1020 # Switch to the new provider.
1021 if self.app_config.state.playing:
1022 self.on_play_pause()
1023 self.app_config.current_provider_id = provider_id
1024 self.app_config.save()
1025 self.update_window(force=True)
1027 dialog.destroy()
1029 def update_window(self, force: bool = False):
1030 if not self.window:
1031 return
1032 logging.info(f"Updating window force={force}")
1033 GLib.idle_add(
1034 lambda: self.window.update(
1035 self.app_config, self.player_manager, force=force
1036 )
1037 )
1039 def update_play_state_from_server(self, prompt_confirm: bool = False):
1040 # TODO (#129): need to make the play queue list loading for the duration here if
1041 # prompt_confirm is False.
1042 if not prompt_confirm and self.app_config.state.playing:
1043 assert self.player_manager
1044 self.player_manager.pause()
1045 self.app_config.state.playing = False
1046 self.app_config.state.loading_play_queue = True
1047 self.update_window()
1049 def do_update(f: Result[PlayQueue]):
1050 play_queue = f.result()
1051 if not play_queue:
1052 self.app_config.state.loading_play_queue = False
1053 return
1054 play_queue.position = play_queue.position or timedelta(0)
1056 new_play_queue = tuple(s.id for s in play_queue.songs)
1057 new_song_progress = play_queue.position
1059 def do_resume(clear_notification: bool):
1060 assert self.player_manager
1061 if was_playing := self.app_config.state.playing:
1062 self.on_play_pause()
1064 self.app_config.state.play_queue = new_play_queue
1065 self.app_config.state.song_progress = play_queue.position
1066 self.app_config.state.current_song_index = play_queue.current_index or 0
1067 self.app_config.state.loading_play_queue = False
1068 self.player_manager.reset()
1069 if clear_notification:
1070 self.app_config.state.current_notification = None
1071 self.update_window()
1073 if was_playing:
1074 self.on_play_pause()
1076 if prompt_confirm:
1077 # If there's not a significant enough difference in the song state,
1078 # don't prompt.
1079 progress_diff = 15.0
1080 if self.app_config.state.song_progress:
1081 progress_diff = abs(
1082 (
1083 self.app_config.state.song_progress - new_song_progress
1084 ).total_seconds()
1085 )
1087 if (
1088 self.app_config.state.play_queue == new_play_queue
1089 and self.app_config.state.current_song
1090 ):
1091 song_index = self.app_config.state.current_song_index
1092 if song_index == play_queue.current_index and progress_diff < 15:
1093 return
1095 # Show a notification to resume the play queue.
1096 resume_text = "Do you want to resume the play queue"
1097 if play_queue.changed_by or play_queue.changed:
1098 resume_text += " saved"
1099 if play_queue.changed_by:
1100 resume_text += f" by {play_queue.changed_by}"
1101 if play_queue.changed:
1102 changed_str = play_queue.changed.astimezone(tz=None).strftime(
1103 "%H:%M on %Y-%m-%d"
1104 )
1105 resume_text += f" at {changed_str}"
1106 resume_text += "?"
1108 self.app_config.state.current_notification = UIState.UINotification(
1109 markup=f"<b>{resume_text}</b>",
1110 actions=(("Resume", partial(do_resume, True)),),
1111 )
1112 self.update_window()
1114 else: # just resume the play queue immediately
1115 do_resume(False)
1117 play_queue_future = AdapterManager.get_play_queue()
1118 play_queue_future.add_done_callback(lambda f: GLib.idle_add(do_update, f))
1120 song_playing_order_token = 0
1121 batch_download_jobs: Set[Result] = set()
1123 def play_song(
1124 self,
1125 song_index: int,
1126 reset: bool = False,
1127 old_play_queue: Tuple[str, ...] = None,
1128 play_queue: Tuple[str, ...] = None,
1129 playable_song_search_direction: int = 1,
1130 ):
1131 def do_reset():
1132 self.player_manager.reset()
1133 self.app_config.state.song_progress = timedelta(0)
1134 self.should_scrobble_song = True
1136 # Tell the player that the next song is available for gapless playback
1137 def do_notify_next_song(next_song: Song):
1138 try:
1139 next_uri = AdapterManager.get_song_file_uri(next_song)
1140 if self.player_manager:
1141 self.player_manager.next_media_cached(next_uri, next_song)
1142 except CacheMissError:
1143 logging.debug(
1144 "Couldn't find the file for next song for gapless playback"
1145 )
1147 # Do this the old fashioned way so that we can have access to ``reset``
1148 # in the callback.
1149 @dbus_propagate(self)
1150 def do_play_song(order_token: int, song: Song):
1151 assert self.player_manager
1152 if order_token != self.song_playing_order_token:
1153 return
1155 uri = None
1156 try:
1157 if "file" in self.player_manager.supported_schemes:
1158 uri = AdapterManager.get_song_file_uri(song)
1159 except CacheMissError:
1160 logging.debug("Couldn't find the file, will attempt to stream.")
1162 if not uri:
1163 try:
1164 uri = AdapterManager.get_song_stream_uri(song)
1165 except Exception:
1166 pass
1167 if (
1168 not uri
1169 or urlparse(uri).scheme not in self.player_manager.supported_schemes
1170 ):
1171 self.app_config.state.current_notification = UIState.UINotification(
1172 markup=f"<b>Unable to play {song.title}.</b>",
1173 icon="dialog-error",
1174 )
1175 return
1177 # Prevent it from doing the thing where it continually loads
1178 # songs when it has to download.
1179 if reset:
1180 do_reset()
1182 # Start playing the song.
1183 if order_token != self.song_playing_order_token:
1184 return
1186 self.player_manager.play_media(
1187 uri,
1188 timedelta(0) if reset else self.app_config.state.song_progress,
1189 song,
1190 )
1191 self.app_config.state.playing = True
1192 self.update_window()
1194 # Check if the next song is available in the cache
1195 if (next_song_index := self.app_config.state.next_song_index) is not None:
1196 next_song_details_future = AdapterManager.get_song_details(
1197 self.app_config.state.play_queue[next_song_index]
1198 )
1200 next_song_details_future.add_done_callback(
1201 lambda f: GLib.idle_add(do_notify_next_song, f.result()),
1202 )
1204 # Show a song play notification.
1205 if self.app_config.song_play_notification:
1206 try:
1207 if glib_notify_exists:
1208 notification_lines = []
1209 if album := song.album:
1210 notification_lines.append(f"<i>{album.name}</i>")
1211 if artist := song.artist:
1212 notification_lines.append(artist.name)
1213 song_notification = Notify.Notification.new(
1214 song.title,
1215 bleach.clean("\n".join(notification_lines)),
1216 )
1217 song_notification.add_action(
1218 "clicked",
1219 "Open Sublime Music",
1220 lambda *a: self.window.present() if self.window else None,
1221 )
1222 song_notification.show()
1224 def on_cover_art_download_complete(cover_art_filename: str):
1225 if order_token != self.song_playing_order_token:
1226 return
1228 # Add the image to the notification, and re-show
1229 # the notification.
1230 song_notification.set_image_from_pixbuf(
1231 GdkPixbuf.Pixbuf.new_from_file_at_scale(
1232 cover_art_filename, 70, 70, True
1233 )
1234 )
1235 song_notification.show()
1237 cover_art_result = AdapterManager.get_cover_art_uri(
1238 song.cover_art, "file"
1239 )
1240 cover_art_result.add_done_callback(
1241 lambda f: on_cover_art_download_complete(f.result())
1242 )
1244 if sys.platform == "darwin":
1245 notification_lines = []
1246 if album := song.album:
1247 notification_lines.append(album.name)
1248 if artist := song.artist:
1249 notification_lines.append(artist.name)
1250 notification_text = "\n".join(notification_lines)
1251 osascript_command = [
1252 "display",
1253 "notification",
1254 f'"{notification_text}"',
1255 "with",
1256 "title",
1257 f'"{song.title}"',
1258 ]
1260 os.system(f"osascript -e '{' '.join(osascript_command)}'")
1261 except Exception:
1262 logging.warning(
1263 "Unable to display notification. Is a notification daemon running?" # noqa: E501
1264 )
1266 # Download current song and prefetch songs. Only do this if the adapter can
1267 # download songs and allow_song_downloads is True and download_on_stream is
1268 # True.
1269 def on_song_download_complete(song_id: str):
1270 if order_token != self.song_playing_order_token:
1271 return
1273 # Hotswap to the downloaded song.
1274 if (
1275 # TODO (#182) allow hotswap if not playing. This requires being able
1276 # to replace the currently playing URI with something different.
1277 self.app_config.state.playing
1278 and self.app_config.state.current_song
1279 and self.app_config.state.current_song.id == song_id
1280 ):
1281 # Switch to the local media if the player can hotswap without lag.
1282 # For example, MPV can is barely noticable whereas there's quite a
1283 # delay with Chromecast.
1284 assert self.player_manager
1285 if self.player_manager.can_start_playing_with_no_latency:
1286 self.player_manager.play_media(
1287 AdapterManager.get_song_file_uri(song),
1288 self.app_config.state.song_progress,
1289 song,
1290 )
1292 # Handle case where a next-song was previously not cached
1293 # but is now available for the player to use
1294 if self.app_config.state.playing:
1295 next_song_index = self.app_config.state.next_song_index
1296 if (
1297 next_song_index is not None
1298 and self.app_config.state.play_queue[next_song_index] == song_id
1299 ):
1300 next_song_details_future = AdapterManager.get_song_details(
1301 song_id
1302 )
1304 next_song_details_future.add_done_callback(
1305 lambda f: GLib.idle_add(do_notify_next_song, f.result()),
1306 )
1308 # Always update the window
1309 self.update_window()
1311 if (
1312 # This only makes sense if the adapter is networked.
1313 AdapterManager.ground_truth_adapter_is_networked()
1314 # Don't download in offline mode.
1315 and not self.app_config.offline_mode
1316 and self.app_config.allow_song_downloads
1317 and self.app_config.download_on_stream
1318 and AdapterManager.can_batch_download_songs()
1319 ):
1320 song_ids = [song.id]
1322 # Add the prefetch songs.
1323 if (
1324 repeat_type := self.app_config.state.repeat_type
1325 ) != RepeatType.REPEAT_SONG:
1326 song_idx = self.app_config.state.play_queue.index(song.id)
1327 is_repeat_queue = RepeatType.REPEAT_QUEUE == repeat_type
1328 prefetch_idxs = []
1329 for i in range(self.app_config.prefetch_amount):
1330 prefetch_idx: int = song_idx + 1 + i
1331 play_queue_len: int = len(self.app_config.state.play_queue)
1332 if is_repeat_queue or prefetch_idx < play_queue_len:
1333 prefetch_idxs.append(
1334 prefetch_idx % play_queue_len # noqa: S001
1335 )
1336 song_ids.extend(
1337 [self.app_config.state.play_queue[i] for i in prefetch_idxs]
1338 )
1340 self.batch_download_jobs.add(
1341 AdapterManager.batch_download_songs(
1342 song_ids,
1343 before_download=lambda _: self.update_window(),
1344 on_song_download_complete=on_song_download_complete,
1345 one_at_a_time=True,
1346 delay=5,
1347 )
1348 )
1350 if old_play_queue:
1351 self.app_config.state.old_play_queue = old_play_queue
1353 if play_queue:
1354 self.app_config.state.play_queue = play_queue
1356 self.app_config.state.current_song_index = song_index
1358 for job in self.batch_download_jobs:
1359 job.cancel()
1361 self.song_playing_order_token += 1
1363 if play_queue:
1364 GLib.timeout_add(
1365 5000,
1366 partial(
1367 self.save_play_queue,
1368 song_playing_order_token=self.song_playing_order_token,
1369 ),
1370 )
1372 # If in offline mode, go to the first song in the play queue after the given
1373 # song that is actually playable.
1374 if self.app_config.offline_mode:
1375 statuses = AdapterManager.get_cached_statuses(
1376 self.app_config.state.play_queue
1377 )
1378 playable_statuses = (
1379 SongCacheStatus.CACHED,
1380 SongCacheStatus.PERMANENTLY_CACHED,
1381 )
1382 can_play = False
1383 current_song_index = self.app_config.state.current_song_index
1385 if statuses[current_song_index] in playable_statuses:
1386 can_play = True
1387 elif self.app_config.state.repeat_type != RepeatType.REPEAT_SONG:
1388 # See if any other songs in the queue are playable.
1389 play_queue_len = len(self.app_config.state.play_queue)
1390 cursor = (
1391 current_song_index + playable_song_search_direction
1392 ) % play_queue_len
1393 for _ in range(play_queue_len): # Don't infinite loop.
1394 if self.app_config.state.repeat_type == RepeatType.NO_REPEAT:
1395 if (
1396 playable_song_search_direction == 1
1397 and cursor < current_song_index
1398 ) or (
1399 playable_song_search_direction == -1
1400 and cursor > current_song_index
1401 ):
1402 # We wrapped around to the end of the play queue without
1403 # finding a song that can be played, and we aren't allowed
1404 # to loop back.
1405 break
1407 # If we find a playable song, stop and play it.
1408 if statuses[cursor] in playable_statuses:
1409 self.play_song(cursor, reset)
1410 return
1412 cursor = (cursor + playable_song_search_direction) % play_queue_len
1414 if not can_play:
1415 # There are no songs that can be played. Show a notification that you
1416 # have to go online to play anything and then don't go further.
1417 if was_playing := self.app_config.state.playing:
1418 self.on_play_pause()
1420 def go_online_clicked():
1421 self.app_config.state.current_notification = None
1422 self.on_go_online()
1423 if was_playing:
1424 self.on_play_pause()
1426 if all(s == SongCacheStatus.NOT_CACHED for s in statuses):
1427 markup = (
1428 "<b>None of the songs in your play queue are cached for "
1429 "offline playback.</b>\nGo online to start playing your queue."
1430 )
1431 else:
1432 markup = (
1433 "<b>None of the remaining songs in your play queue are cached "
1434 "for offline playback.</b>\nGo online to contiue playing your "
1435 "queue."
1436 )
1438 self.app_config.state.current_notification = UIState.UINotification(
1439 icon="cloud-offline-symbolic",
1440 markup=markup,
1441 actions=(("Go Online", go_online_clicked),),
1442 )
1443 if reset:
1444 do_reset()
1445 self.update_window()
1446 return
1448 song_details_future = AdapterManager.get_song_details(
1449 self.app_config.state.play_queue[self.app_config.state.current_song_index]
1450 )
1451 if song_details_future.data_is_available:
1452 song_details_future.add_done_callback(
1453 lambda f: do_play_song(self.song_playing_order_token, f.result())
1454 )
1455 else:
1456 song_details_future.add_done_callback(
1457 lambda f: GLib.idle_add(
1458 partial(do_play_song, self.song_playing_order_token), f.result()
1459 ),
1460 )
1462 def save_play_queue(self, song_playing_order_token: int = None):
1463 if (
1464 len(self.app_config.state.play_queue) == 0
1465 or self.app_config.provider is None
1466 or (
1467 song_playing_order_token
1468 and song_playing_order_token != self.song_playing_order_token
1469 )
1470 ):
1471 return
1473 position = self.app_config.state.song_progress
1474 self.last_play_queue_update = position or timedelta(0)
1476 if AdapterManager.can_save_play_queue() and self.app_config.state.current_song:
1477 AdapterManager.save_play_queue(
1478 song_ids=self.app_config.state.play_queue,
1479 current_song_index=self.app_config.state.current_song_index,
1480 position=position,
1481 )