from functools import partial
from typing import Any, List, Optional, Tuple, cast
import bleach
from gi.repository import Gdk, Gio, GLib, GObject, Gtk, Pango
from ..adapters import AdapterManager, CacheMissError, Result, api_objects as API
from ..config import AppConfiguration
from ..ui import util
from ..ui.common import IconButton, LoadError, SongListColumn
[docs]class BrowsePanel(Gtk.Overlay):
"""Defines the arist panel."""
__gsignals__ = {
"song-clicked": (
GObject.SignalFlags.RUN_FIRST,
GObject.TYPE_NONE,
(int, object, object),
),
"refresh-window": (
GObject.SignalFlags.RUN_FIRST,
GObject.TYPE_NONE,
(object, bool),
),
}
update_order_token = 0
[docs] def __init__(self):
super().__init__()
scrolled_window = Gtk.ScrolledWindow()
window_box = Gtk.Box()
self.error_container = Gtk.Box()
window_box.pack_start(self.error_container, True, True, 0)
self.root_directory_listing = ListAndDrilldown()
self.root_directory_listing.connect(
"song-clicked",
lambda _, *args: self.emit("song-clicked", *args),
)
self.root_directory_listing.connect(
"refresh-window",
lambda _, *args: self.emit("refresh-window", *args),
)
window_box.add(self.root_directory_listing)
scrolled_window.add(window_box)
self.add(scrolled_window)
self.spinner = Gtk.Spinner(
name="browse-spinner",
active=True,
halign=Gtk.Align.CENTER,
valign=Gtk.Align.CENTER,
)
self.add_overlay(self.spinner)
[docs] def update(self, app_config: AppConfiguration, force: bool = False):
self.update_order_token += 1
def do_update(update_order_token: int, id_stack: Tuple[str, ...]):
if self.update_order_token != update_order_token:
return
if len(id_stack) == 0:
self.root_directory_listing.hide()
if len(self.error_container.get_children()) == 0:
load_error = LoadError(
"Directory list",
"browse to song",
has_data=False,
offline_mode=app_config.offline_mode,
)
self.error_container.pack_start(load_error, True, True, 0)
self.error_container.show_all()
else:
for c in self.error_container.get_children():
self.error_container.remove(c)
self.error_container.hide()
self.root_directory_listing.update(id_stack, app_config, force)
self.spinner.hide()
def calculate_path() -> Tuple[str, ...]:
if (current_dir_id := app_config.state.selected_browse_element_id) is None:
return ("root",)
id_stack = []
while current_dir_id:
try:
directory = AdapterManager.get_directory(
current_dir_id,
before_download=self.spinner.show,
).result()
except CacheMissError as e:
directory = cast(API.Directory, e.partial_data)
if not directory:
break
else:
id_stack.append(directory.id)
current_dir_id = directory.parent_id
return tuple(id_stack)
path_result: Result[Tuple[str, ...]] = Result(calculate_path)
path_result.add_done_callback(
lambda f: GLib.idle_add(partial(do_update, self.update_order_token), f.result())
)
[docs]class ListAndDrilldown(Gtk.Paned):
__gsignals__ = {
"song-clicked": (
GObject.SignalFlags.RUN_FIRST,
GObject.TYPE_NONE,
(int, object, object),
),
"refresh-window": (
GObject.SignalFlags.RUN_FIRST,
GObject.TYPE_NONE,
(object, bool),
),
}
[docs] def __init__(self):
Gtk.Paned.__init__(self, orientation=Gtk.Orientation.HORIZONTAL)
self.list = MusicDirectoryList()
self.list.connect(
"song-clicked",
lambda _, *args: self.emit("song-clicked", *args),
)
self.list.connect(
"refresh-window",
lambda _, *args: self.emit("refresh-window", *args),
)
self.pack1(self.list, False, False)
self.box = Gtk.Box()
self.pack2(self.box, True, False)
[docs] def update(
self,
id_stack: Tuple[str, ...],
app_config: AppConfiguration,
force: bool = False,
):
*child_id_stack, dir_id = id_stack
selected_id = child_id_stack[-1] if len(child_id_stack) > 0 else None
self.show()
self.list.update(
directory_id=dir_id,
selected_id=selected_id,
app_config=app_config,
force=force,
)
children = self.box.get_children()
if len(child_id_stack) == 0:
if len(children) > 0:
self.box.remove(children[0])
else:
if len(children) == 0:
drilldown = ListAndDrilldown()
drilldown.connect(
"song-clicked",
lambda _, *args: self.emit("song-clicked", *args),
)
drilldown.connect(
"refresh-window",
lambda _, *args: self.emit("refresh-window", *args),
)
self.box.add(drilldown)
self.box.show_all()
self.box.get_children()[0].update(tuple(child_id_stack), app_config, force=force)
[docs]class MusicDirectoryList(Gtk.Box):
__gsignals__ = {
"song-clicked": (
GObject.SignalFlags.RUN_FIRST,
GObject.TYPE_NONE,
(int, object, object),
),
"refresh-window": (
GObject.SignalFlags.RUN_FIRST,
GObject.TYPE_NONE,
(object, bool),
),
}
update_order_token = 0
directory_id: Optional[str] = None
selected_id: Optional[str] = None
offline_mode = False
[docs] class DrilldownElement(GObject.GObject):
id = GObject.Property(type=str)
name = GObject.Property(type=str)
[docs] def __init__(self, element: API.Directory):
GObject.GObject.__init__(self)
self.id = element.id
self.name = element.name
[docs] def __init__(self):
Gtk.Box.__init__(self, orientation=Gtk.Orientation.VERTICAL)
list_actions = Gtk.ActionBar()
self.refresh_button = IconButton("view-refresh-symbolic", "Refresh folder")
self.refresh_button.connect("clicked", lambda *a: self.update(force=True))
list_actions.pack_end(self.refresh_button)
self.add(list_actions)
self.loading_indicator = Gtk.ListBox()
spinner_row = Gtk.ListBoxRow(activatable=False, selectable=False)
spinner = Gtk.Spinner(name="drilldown-list-spinner", active=True)
spinner_row.add(spinner)
self.loading_indicator.add(spinner_row)
self.pack_start(self.loading_indicator, False, False, 0)
self.error_container = Gtk.Box()
self.add(self.error_container)
self.scroll_window = Gtk.ScrolledWindow(min_content_width=250)
scrollbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
self.drilldown_directories_store = Gio.ListStore()
self.list = Gtk.ListBox()
self.list.bind_model(self.drilldown_directories_store, self.create_row)
scrollbox.add(self.list)
# clickable, cache status, title, duration, song ID
self.directory_song_store = Gtk.ListStore(bool, str, str, str, str)
self.directory_song_list = Gtk.TreeView(
model=self.directory_song_store,
name="directory-songs-list",
headers_visible=False,
)
selection = self.directory_song_list.get_selection()
selection.set_mode(Gtk.SelectionMode.MULTIPLE)
selection.set_select_function(lambda _, model, path, current: model[path[0]][0])
# Song status column.
renderer = Gtk.CellRendererPixbuf()
renderer.set_fixed_size(30, 35)
column = Gtk.TreeViewColumn("", renderer, icon_name=1)
column.set_resizable(True)
self.directory_song_list.append_column(column)
self.directory_song_list.append_column(SongListColumn("TITLE", 2, bold=True))
self.directory_song_list.append_column(SongListColumn("DURATION", 3, align=1, width=40))
self.directory_song_list.connect("row-activated", self.on_song_activated)
self.directory_song_list.connect("button-press-event", self.on_song_button_press)
scrollbox.add(self.directory_song_list)
self.scroll_window.add(scrollbox)
self.pack_start(self.scroll_window, True, True, 0)
[docs] def update(
self,
app_config: AppConfiguration | None = None,
force: bool = False,
directory_id: str | None = None,
selected_id: str | None = None,
):
self.directory_id = directory_id or self.directory_id
self.selected_id = selected_id or self.selected_id
self.update_store(
self.directory_id,
force=force,
order_token=self.update_order_token,
)
if app_config:
# Deselect everything if switching online to offline.
if self.offline_mode != app_config.offline_mode:
self.directory_song_list.get_selection().unselect_all()
for c in self.error_container.get_children():
self.error_container.remove(c)
self.offline_mode = app_config.offline_mode
self.refresh_button.set_sensitive(not self.offline_mode)
_current_child_ids: List[str] = []
[docs] @util.async_callback(
AdapterManager.get_directory,
before_download=lambda self: self.loading_indicator.show(),
on_failure=lambda self, e: self.loading_indicator.hide(),
)
def update_store(
self,
directory: API.Directory,
app_config: AppConfiguration | None = None,
force: bool = False,
order_token: int | None = None,
is_partial: bool = False,
):
if order_token != self.update_order_token:
return
dir_children = directory.children or []
for c in self.error_container.get_children():
self.error_container.remove(c)
if is_partial:
load_error = LoadError(
"Directory listing",
"load directory",
has_data=len(dir_children) > 0,
offline_mode=self.offline_mode,
)
self.error_container.pack_start(load_error, True, True, 0)
self.error_container.show_all()
else:
self.error_container.hide()
# This doesn't look efficient, since it's doing a ton of passses over the data,
# but there is some annoying memory overhead for generating the stores to diff,
# so we are short-circuiting by checking to see if any of the the IDs have
# changed.
#
# The entire algorithm ends up being O(2n), but the first loop is very tight,
# and the expensive parts of the second loop are avoided if the IDs haven't
# changed.
children_ids, children, song_ids = [], [], []
selected_dir_idx = None
if len(self._current_child_ids) != len(dir_children):
force = True
for i, c in enumerate(dir_children):
if i >= len(self._current_child_ids) or c.id != self._current_child_ids[i]:
force = True
if c.id == self.selected_id:
selected_dir_idx = i
children_ids.append(c.id)
children.append(c)
if not hasattr(c, "children"):
song_ids.append(c.id)
if force:
new_directories_store = []
self._current_child_ids = children_ids
songs = []
for el in children:
if hasattr(el, "children"):
new_directories_store.append(
MusicDirectoryList.DrilldownElement(cast(API.Directory, el))
)
else:
songs.append(cast(API.Song, el))
util.diff_model_store(self.drilldown_directories_store, new_directories_store)
def song_sort_key(song: API.Song) -> Tuple[Optional[int], Optional[int]]:
return (
song.disc_number if hasattr(song, "disc_number") else 0,
song.track if hasattr(song, "track") else 0,
)
songs.sort(key=song_sort_key)
new_songs_store = [
[
(
not self.offline_mode
or status_icon in ("folder-download-symbolic", "view-pin-symbolic")
),
status_icon,
bleach.clean(song.title),
util.format_song_duration(song.duration),
song.id,
]
for status_icon, song in zip(util.get_cached_status_icons(song_ids), songs)
]
else:
new_songs_store = [
[
(
not self.offline_mode
or status_icon in ("folder-download-symbolic", "view-pin-symbolic")
),
status_icon,
*song_model[2:],
]
for status_icon, song_model in zip(
util.get_cached_status_icons(song_ids), self.directory_song_store
)
]
util.diff_song_store(self.directory_song_store, new_songs_store)
self.directory_song_list.show()
if len(self.drilldown_directories_store) == 0:
self.list.hide()
else:
self.list.show()
if len(self.directory_song_store) == 0:
self.directory_song_list.hide()
self.scroll_window.set_min_content_width(275)
else:
self.directory_song_list.show()
self.scroll_window.set_min_content_width(350)
# Preserve selection
if selected_dir_idx is not None:
row = self.list.get_row_at_index(selected_dir_idx)
self.list.select_row(row)
self.loading_indicator.hide()
[docs] def on_download_state_change(self, _):
GLib.idle_add(self.update)
# Create Element Helper Functions
# ==================================================================================
[docs] def create_row(self, model: DrilldownElement) -> Gtk.ListBoxRow:
row = Gtk.ListBoxRow(
action_name="app.browse-to",
action_target=GLib.Variant("s", model.id),
)
rowbox = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL)
rowbox.add(
Gtk.Label(
label=bleach.clean(f"<b>{model.name}</b>"),
use_markup=True,
margin=8,
halign=Gtk.Align.START,
ellipsize=Pango.EllipsizeMode.END,
)
)
image = Gtk.Image.new_from_icon_name("go-next-symbolic", Gtk.IconSize.BUTTON)
rowbox.pack_end(image, False, False, 5)
row.add(rowbox)
row.show_all()
return row
# Event Handlers
# ==================================================================================
[docs] def on_song_activated(self, treeview: Any, idx: Gtk.TreePath, column: Any):
if not self.directory_song_store[idx[0]][0]:
return
# The song ID is in the last column of the model.
self.emit(
"song-clicked",
idx.get_indices()[0],
[m[-1] for m in self.directory_song_store],
{},
)