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()