playtypus/mainwindow.py

# -*- coding: utf-8 -*-
import requests
from funksnake import Funkwhale
from PyQt5 import QtCore, QtGui, QtWidgets, QtNetwork

from ui_mainwindow import UiMainWindow
from settingsdialog import SettingsDialog


class ThreadWorker(QtCore.QObject):
    """
    The :class:`~threads.ThreadWorker` class provides a base class for thread workers.

    In Qt5, all work is done in workers, rather than actual QThread classes, and this case class provides
    some utility signals and slots to make it easier to use threads and thread workers.
    """
    quit = QtCore.pyqtSignal()
    error = QtCore.pyqtSignal(str, str)

    def start(self):
        """
        The start method is how the worker runs. Basically, put your code here.
        """
        raise NotImplementedError('Your class needs to override this method and run self.quit.emit() when complete')


class AlbumArtWorker(ThreadWorker):
    album_item = None
    artwork_url = None

    def start(self):
        if not self.album_item:
            self.error.emit('No album QListWidgetItem set', 'The album_item attribute was not set')
            self.quit.emit()
            return
        if not self.artwork_url:
            self.error.emit('No artwork URL set', 'The artwork_url attribute was not set')
            self.quit.emit()
            return
        response = requests.get(self.artwork_url)
        if response.status_code == 200:
            pixmap = QtGui.QPixmap()
            pixmap.loadFromData(response.content)
            self.album_item.setIcon(QtGui.QIcon(pixmap))
        else:
            print(response.text)
        self.quit.emit()


class AlbumWorker(ThreadWorker):
    """
    A thread worker to fetch the album art
    """
    window = None

    def start(self):
        if not self.window:
            self.error.emit('No window set', 'The window attribute was not set')
            self.quit.emit()
            return
        self.window.albumListWidget.clear()
        for album in self.window.funkwhale.albums.list()['results']:
            album_item = QtWidgets.QListWidgetItem(album['title'])
            self.window.albumListWidget.addItem(album_item)
            artwork_worker = AlbumArtWorker()
            artwork_worker.album_item = album_item
            artwork_worker.artwork_url = album['cover']['urls']['medium_square_crop']
            self.window.run_thread(artwork_worker, 'album-{}'.format(album['title']))
        self.quit.emit()


class MainWindow(QtWidgets.QMainWindow, UiMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setup_ui()
        self.settingsAction.triggered.connect(self.on_settings_action_triggered)
        self.settings = QtCore.QSettings('info.snyman.pyfunkwhale', 'PyFunkwhale')
        self.network_manager = QtNetwork.QNetworkAccessManager()
        self.settings_dialog = SettingsDialog(self)
        self.funkwhale = None
        self.threads = {}
        if self.settings.contains('funkwhale/server_url'):
            self.setup_funkwhale(self.settings.value('funkwhale/server_url'),
                                 self.settings.value('funkwhale/username', None),
                                 self.settings.value('funkwhale/password', None))
            self.load_albums()
            # self.load_artists()

    def run_thread(self, worker, thread_name):
        """
        Create a thread and its worker. This gets rid of a lot of unnecessary boilerplate code.

        :param ThreadWorker worker: A worker object that actually performs the work
        :param str thread_name: The name of the thread, used to keep track of the thread.
        :param bool can_start: Start the thread. Defaults to True.
        """
        if not thread_name:
            raise ValueError('A thread name is required')
        if thread_name in self.threads:
            raise KeyError('A thread with name "{}" already exists'.format(thread_name))
        # Create the thread and add the thread and the worker to the thread tracker
        thread = QtCore.QThread()
        self.threads[thread_name] = {
            'thread': thread,
            'worker': worker
        }
        # Move the worker into the thread's context
        worker.moveToThread(thread)
        # Connect slots and signals
        thread.started.connect(worker.start)
        worker.quit.connect(thread.quit)
        worker.quit.connect(worker.deleteLater)
        thread.finished.connect(thread.deleteLater)
        thread.finished.connect(self.make_remove_thread(thread_name))
        thread.start()

    def get_thread_worker(self, thread_name):
        """
        Get the worker by the thread name

        :param str thread_name: The name of the thread
        :returns: The worker for this thread name, or None
        """
        thread = self.threads.get(thread_name)
        if not thread:
            return
        return thread.get('worker')

    def is_thread_finished(self, thread_name):
        """
        Check if a thread is finished running.

        :param str thread_name: The name of the thread
        :returns: True if the thread is finished, False if it is still running
        """
        return thread_name not in self.threads or self.threads[thread_name]['thread'].isFinished()

    def make_remove_thread(self, thread_name):
        """
        Create a function to remove the thread once the thread is finished.

        :param str thread_name: The name of the thread which should be removed from the thread registry.
        :returns: A function which will remove the thread from the thread registry.
        """
        def remove_thread():
            """
            Stop and remove a registered thread

            :param str thread_name: The name of the thread to stop and remove
            """
            if thread_name in self.threads:
                del self.threads[thread_name]

        return remove_thread

    def setup_funkwhale(self, server_url=None, username=None, password=None):
        if not self.funkwhale and not server_url:
            raise ValueError('A server url needs to be defined')
        if not self.funkwhale and server_url:
            self.funkwhale = Funkwhale(server_url)
        if username and password:
            self.funkwhale.login(username, password)

    def load_albums(self):
        if not self.funkwhale or not self.funkwhale.token:
            return
        worker = AlbumWorker()
        worker.window = self
        self.run_thread(worker, 'get-albums')

    def load_artists(self):
        if not self.funkwhale or not self.funkwhale.token:
            return
        self.artistListWidget.clear()
        for artist in self.funkwhale.artists.list()['results']:
            self.artistListWidget.addItem(QtWidgets.QListWidgetItem(artist['name']))

    def on_settings_action_triggered(self):
        self.settings_dialog.server_url = self.settings.value('funkwhale/server_url', '')
        self.settings_dialog.username = self.settings.value('funkwhale/username', '')
        self.settings_dialog.password = self.settings.value('funkwhale/password', '')
        if self.settings_dialog.exec() == QtWidgets.QDialog.Accepted:
            self.settings.setValue('funkwhale/server_url', self.settings_dialog.server_url)
            self.settings.setValue('funkwhale/username', self.settings_dialog.username)
            self.settings.setValue('funkwhale/password', self.settings_dialog.password)
            self.setup_funkwhale(self.settings.value('funkwhale/server_url'),
                                 self.settings.value('funkwhale/username', None),
                                 self.settings.value('funkwhale/password', None))