import uuid import time from pathlib import Path from typing import Optional, Callable, Dict, List, Any import bleach from gi.repository import Gdk, Gtk, Handy, GLib, Pango from ..adapters import ( AdapterManager, api_objects as API, DownloadProgress, Result, ConfigurationStore, ConfigParamDescriptor, ) from ..config import AppConfiguration, ProviderConfiguration from ..players import PlayerManager from .actions import run_action, variant_type_from_python from .common import IconButton class ProvidersWindow(Handy.Window): is_initial = True def __init__(self, main_window): Handy.Window.__init__(self, modal=True, window_position=Gtk.WindowPosition.CENTER_ON_PARENT, destroy_with_parent=True, type_hint=Gdk.WindowTypeHint.DIALOG, default_width=640, default_height=576) self.main_window = main_window # Don't die when closed def hide_not_destroy(*_): if self.is_initial: run_action(self.main_window, 'app.quit') return True self.hide() return True self.connect('delete-event', hide_not_destroy) self.stack = Gtk.Stack() self.create_page = CreateProviderPage(self) self.stack.add(self.create_page) self.configure_page = ConfigureProviderPage(self) self.stack.add(self.configure_page) self.status_page = ProviderStatusPage(self) self.stack.add(self.status_page) self.add(self.stack) def update(self, app_config: AppConfiguration, player_manager: PlayerManager): self.is_initial = app_config.current_provider_id is None self.status_page.update(app_config, player_manager) def _set_transition(self, going_back: bool): if not self.is_visible(): self.stack.set_transition_type(Gtk.StackTransitionType.NONE) return if going_back: self.stack.set_transition_type(Gtk.StackTransitionType.SLIDE_RIGHT) else: self.stack.set_transition_type(Gtk.StackTransitionType.SLIDE_LEFT) def open_create_page(self, go_back: Optional[Callable] = None, going_back=False): self._set_transition(going_back) self.create_page.setup(go_back) self.create_page.show() self.stack.set_visible_child(self.create_page) def open_configure_page(self, id, adapter, config, go_back: Callable): self._set_transition(False) self.configure_page.setup(id, adapter, config, go_back) self.configure_page.show() self.stack.set_visible_child(self.configure_page) def open_status_page(self): self._set_transition(True) self.status_page.show() self.stack.set_visible_child(self.status_page) class CreateProviderPage(Gtk.Box): def __init__(self, window): Gtk.Box.__init__(self, orientation=Gtk.Orientation.VERTICAL) self.window = window header_bar = Handy.HeaderBar(title="Add Music Source") self.cancel_button = Gtk.Button(label='Quit') self.cancel_button.connect('clicked', self._on_cancel_clicked) header_bar.pack_start(self.cancel_button) self.add(header_bar) scrolled_window = Gtk.ScrolledWindow() scrolled_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) list_box = Gtk.ListBox(expand=True, selection_mode=Gtk.SelectionMode.NONE) adapters = AdapterManager.available_adapters adapters = sorted( filter(lambda a: a.can_be_ground_truth, adapters), key=lambda a: a.get_ui_info().name) for adapter in adapters: ui_info: UIInfo = adapter.get_ui_info() row = Handy.ActionRow( title=ui_info.name, icon_name=ui_info.icon_name(), subtitle=ui_info.description, subtitle_lines=4, use_underline=True, activatable=True) def activated(*_, adapter=adapter): self.window.open_configure_page( None, adapter, ConfigurationStore(), lambda: self.window.open_create_page(self._go_back, going_back=True)) row.connect('activated', activated) row.add(Gtk.Image(icon_name='go-next-symbolic')) list_box.add(row) scrolled_box.add(list_box) scrolled_window.add(scrolled_box) self.pack_start(scrolled_window, True, True, 0) def setup(self, go_back: Optional[Callable]): self._go_back = go_back if go_back: self.cancel_button.set_label("Cancel") else: self.cancel_button.set_label("Quit") def _on_cancel_clicked(self, *_): if self._go_back: self._go_back() else: self.window.close() class ConfigureProviderPage(Gtk.Box): _id = None _adapter = None _go_back = None _config_store = None _config_widgets = {} _config_updates = {} _required_fields: Dict[str, Callable[[Any], Optional[str]]] = {} _errors = {} _validation_ratchet = 0 _had_all_required = False def __init__(self, window): Gtk.Box.__init__(self, orientation=Gtk.Orientation.VERTICAL) self.window = window self.header_bar = Handy.HeaderBar(show_close_button=False, title="Configuration") back_button = Gtk.Button(label="Back") back_button.connect('clicked', lambda *_: self._go_back()) self.header_bar.pack_start(back_button) self.create_button = Gtk.Button(label="Create", sensitive=False) self.create_button.get_style_context().add_class('suggested-action') self.create_button.connect('clicked', self._on_create_clicked) self.header_bar.pack_end(self.create_button) self.add(self.header_bar) self.status_revealer = Gtk.Revealer(reveal_child=False) box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10, margin=10) clamp = Handy.Clamp() self.status_stack = Gtk.Stack() self.status_spinner = Gtk.Spinner() self.status_stack.add(self.status_spinner) self.status_ok_image = Gtk.Image(icon_name="config-ok-symbolic") self.status_stack.add(self.status_ok_image) self.status_error_image = Gtk.Image(icon_name="config-error-symbolic") self.status_stack.add(self.status_error_image) box.pack_start(self.status_stack, False, True, 0) self.status_label = Gtk.Label(ellipsize=Pango.EllipsizeMode.END, lines=4, justify=Gtk.Justification.LEFT, wrap=True) box.pack_start(self.status_label, False, True, 0) clamp.add(box) self.status_revealer.add(clamp) self.pack_start(self.status_revealer, False, True, 0) scrolled_window = Gtk.ScrolledWindow() clamp = Handy.Clamp(margin=12) self.settings_list = Gtk.ListBox() self.settings_list.get_style_context().add_class('content') clamp.add(self.settings_list) scrolled_window.add(clamp) self.pack_start(scrolled_window, True, True, 0) def setup(self, id_, adapter, config_store, go_back): self._id = id_ self._adapter = adapter self._go_back = go_back # Reset self._config_store = config_store self._config_updates = {} self._required_fields = {} self._errors = {} self._validation_ratchet = 0 self._had_all_required = False self.create_button.set_sensitive(False) if self._id is not None: self.create_button.set_label('Save') else: self.create_button.set_label('Create') for child in self.settings_list.get_children(): self.settings_list.remove(child) name = adapter.get_ui_info().name self.header_bar.set_title(f'{name} Configuration') # Reset status revealer self.status_revealer.set_reveal_child(False) # First row is always the name name_row = self._create_entry_row('name', ConfigParamDescriptor(str, 'Name')) if 'name' in self._config_store: self._config_updates['name'](self._config_store['name']) self.settings_list.add(name_row) # Collected advanced settings in an expander row advanced_row = Handy.ExpanderRow(title="Advanced", expanded=False) self._params, self._validate = adapter.get_configuration_form() for name, config in self._params.items(): row = { str: self._create_entry_row, bool: self._create_switch_row, int: self._create_spin_button_row, 'password': lambda *a: self._create_entry_row(*a, is_password=True), 'option': self._create_option_row, Path: self._create_path_row, }[config.type](name, config) # Set the initial value from the config store if name in self._config_store: self._config_updates[name](self._config_store[name]) elif config.default is not None: self._config_store[name] = config.default if config.advanced: advanced_row.add(row) else: self.settings_list.add(row) self.settings_list.add(advanced_row) self.show_all() def _create_entry_row(self, name, config, is_password=False): row = Handy.ActionRow(title=config.description, subtitle=config.helptext, subtitle_lines=4, can_focus=False, selectable=False) entry = Gtk.Entry( valign=Gtk.Align.CENTER, visibility=not is_password, input_purpose=Gtk.InputPurpose.PASSWORD if is_password else Gtk.InputPurpose.FREE_FORM) if config.default is not None: if is_password: entry.set_text(config.default[1]) else: entry.set_text(config.default) if config.required: if is_password: self._required_fields[name] = lambda v: len(v[1]) > 0 else: self._required_fields[name] = lambda v: len(v) > 0 entry.connect('notify::text', lambda *_: self._update_config(name, entry.get_text(), is_password)) def set_value(value): if is_password: if value[0] != 'plaintext': return value = value[1] entry.set_text(value) self._config_updates[name] = set_value self._config_widgets[name] = entry row.add(entry) return row def _create_switch_row(self, name, config): row = Handy.ActionRow(title=config.description, subtitle=config.helptext, subtitle_lines=4, can_focus=False, selectable=False) switch = Gtk.Switch(valign=Gtk.Align.CENTER) assert config.default is not None switch.set_active(config.default) switch.connect('notify::active', lambda *_: self._update_config(name, switch.get_active())) self._config_updates[name] = switch.set_active self._config_widgets[name] = switch row.add(switch) return row def _create_spin_button_row(self, name, config): pass def _create_option_row(self, name, config): pass def _create_path_row(self, name, config): pass def _verify_required(self): errors = {} for name, verify in self._required_fields.items(): if name not in self._config_store or not verify(self._config_store[name]): errors[name] = 'Missing field' return errors def _update_config(self, name, value, is_secret=False): if is_secret: self._config_store.set_secret(name, value) else: self._config_store[name] = value # Reset errors for name in self._errors.keys(): widget = self._config_widgets[name] widget.get_style_context().remove_class('error') widget.set_tooltip_markup(None) self._errors = {} self.create_button.set_sensitive(False) if not self._had_all_required and self._verify_required(): return self.status_revealer.set_reveal_child(True) self._update_status(True, '') self._had_all_required = True self._validation_ratchet += 1 ratchet = self._validation_ratchet def on_verify_result(errors: Optional[Dict[str, str]]): if errors is None: return errors.update(self._verify_required()) # Update controls after validation, as fields may have changed for name, update in self._config_updates.items(): if name in self._config_store: update(self._config_store[name]) self._errors = errors self.create_button.set_sensitive(len(errors) == 0) if '__ping__' in self._errors: ping_error = self._errors.pop('__ping__') else: ping_error = None for name, error in self._errors.items(): widget = self._config_widgets[name] widget.get_style_context().add_class('error') widget.set_tooltip_markup(error) self._update_status(False, ping_error) def validate(): time.sleep(0.75) if self._validation_ratchet != ratchet: return None return self._validate(self._config_store) result = Result(validate) result.add_done_callback( lambda f: GLib.idle_add(on_verify_result, f.result()) ) def _update_status(self, verifying: bool, error: Optional[str]): if verifying: self.status_spinner.start() self.status_stack.set_visible_child(self.status_spinner) self.status_label.set_markup('Verifying Connection...') elif error: self.status_stack.set_visible_child(self.status_error_image) self.status_label.set_markup(bleach.clean(error)) else: self.status_stack.set_visible_child(self.status_ok_image) self.status_label.set_markup('Connected Successfully') if not verifying: self.status_spinner.stop() def _on_create_clicked(self, *_): id = self._id or str(uuid.uuid4()) name = self._config_store.pop('name') config = {} for key, item in self._config_store.items(): type = self._params[key].type if type == 'password': type = List[str] elif type == 'option': type = str config[key] = GLib.Variant(variant_type_from_python(type), item) adapter_name = self._adapter.get_ui_info().name run_action( self.window.main_window, 'providers.set-config', id, name, adapter_name, config) GLib.idle_add(lambda *_: self.window.close()) class ProviderStatusPage(Gtk.Box): _current_provider = None def __init__(self, window): Gtk.Box.__init__(self, orientation=Gtk.Orientation.VERTICAL) self.window = window header_bar = Handy.HeaderBar(show_close_button=True, title="Connection Status") self.add(header_bar) scrolled_window = Gtk.ScrolledWindow() clamp = Handy.Clamp(margin=12) box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10) self.title = Gtk.Label(wrap=True, justify=Gtk.Justification.LEFT) self.title.get_style_context().add_class('title') self.title.get_style_context().add_class('large-title') box.add(self.title) self.status_stack = Gtk.Stack() self.status_offline = self._create_status_box('offline', 'Offline') self.status_stack.add(self.status_offline) self.status_connected = self._create_status_box('connected', 'Connected') self.status_stack.add(self.status_connected) self.status_error = self._create_status_box('error', 'Error Connecting to Server') self.status_stack.add(self.status_error) box.add(self.status_stack) list_box = Gtk.ListBox() list_box.get_style_context().add_class('content') row = Handy.ActionRow(title='Offline Mode', can_focus=False, selectable=False) self.offline_switch = Gtk.Switch(valign=Gtk.Align.CENTER) row.add(self.offline_switch) list_box.add(row) row = Handy.ActionRow( title='Edit Configuration...', use_underline=True, activatable=True) def activated(*_): config = self._current_provider.ground_truth_adapter_config.clone() config['name'] = self._current_provider.name self.window.open_configure_page( self._current_provider.id, self._current_provider.ground_truth_adapter_type, config, lambda: self.window.open_status_page()) row.connect('activated', activated) row.add(Gtk.Image(icon_name='go-next-symbolic')) list_box.add(row) self.provider_list = Handy.ExpanderRow(title="Other Providers", expanded=False) add_button = IconButton(icon_name='list-add-symbolic', valign=Gtk.Align.CENTER, relief=True) def add_clicked(*_): self.window.open_create_page(lambda: self.window.open_status_page()) add_button.connect('clicked', add_clicked) self.provider_list.add_action(add_button) list_box.add(self.provider_list) box.pack_start(list_box, False, True, 10) clamp.add(box) scrolled_window.add(clamp) self.pack_start(scrolled_window, True, True, 0) def _create_status_box(self, icon: str, label: str): box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10, halign=Gtk.Align.CENTER) icon = Gtk.Image(icon_name=f"server-{icon}-symbolic") box.add(icon) label = Gtk.Label(label=label) box.add(label) return box _other_providers_cache = None _other_providers = [] def update(self, app_config: AppConfiguration, player_manager: PlayerManager): assert AdapterManager.ground_truth_adapter_is_networked self._current_provider = app_config.provider self.title.set_label(app_config.provider.name) if app_config.offline_mode: self.status_stack.set_visible_child(self.status_offline) elif AdapterManager.get_ping_status(): self.status_stack.set_visible_child(self.status_connected) else: self.status_stack.set_visible_child(self.status_error) other_providers = [id for id in app_config.providers.keys() if id != app_config.current_provider_id] if self._other_providers_cache is None or self._other_providers_cache != other_providers: self._other_providers_cache = other_providers for child in self._other_providers: self.provider_list.remove(child) self._other_providers = [] self.provider_list.set_enable_expansion(len(other_providers) > 0) for id in other_providers: provider = app_config.providers[id] row = Handy.ActionRow(title=provider.name, can_focus=False, selectable=False) button = Gtk.Button(label="Switch", valign=Gtk.Align.CENTER) def on_clicked(*_, id=id): run_action(self.window.main_window, 'providers.switch', id) button.connect('clicked', on_clicked) row.add(button) button = IconButton(icon_name='user-trash-symbolic', valign=Gtk.Align.CENTER, relief=True) button.get_style_context().add_class('destructive-action') def on_clicked(*_, id=id): run_action(self.window.main_window, 'providers.remove', id) button.connect('clicked', on_clicked) row.add(button) self.provider_list.add(row) self._other_providers.append(row) self.provider_list.show_all()