# -*- 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))