4e1248a18a
We don't need to know if Origin is installed before launching the window, so we can save on startup time by executing the worker after the window has become visible.
96 lines
2.4 KiB
Python
96 lines
2.4 KiB
Python
import sys
|
|
from abc import abstractmethod
|
|
from dataclasses import dataclass
|
|
from enum import IntEnum
|
|
from typing import Optional
|
|
|
|
from PyQt5.QtCore import QRunnable, QObject, pyqtSlot, pyqtSignal
|
|
|
|
|
|
class Worker(QRunnable):
|
|
"""
|
|
Base QRunnable class.
|
|
|
|
This class provides a base for QRunnables with signals that are automatically deleted.
|
|
|
|
To use this class you have to assign the signals object of your concrete implementation
|
|
to the `Worker.signals` attribute and implement `Worker.run_real()`
|
|
"""
|
|
|
|
def __init__(self):
|
|
super(Worker, self).__init__()
|
|
self.setAutoDelete(True)
|
|
self.__signals: Optional[QObject] = None
|
|
|
|
@property
|
|
def signals(self) -> QObject:
|
|
if self.__signals is None:
|
|
raise RuntimeError(f"Subclasses must implement '{type(self).__name__}.signals' QObject attribute")
|
|
return self.__signals
|
|
|
|
@signals.setter
|
|
def signals(self, obj: QObject):
|
|
self.__signals = obj
|
|
|
|
@abstractmethod
|
|
def run_real(self):
|
|
pass
|
|
|
|
@pyqtSlot()
|
|
def run(self):
|
|
self.run_real()
|
|
self.signals.deleteLater()
|
|
|
|
|
|
class QueueWorkerState(IntEnum):
|
|
UNDEFINED = 0
|
|
QUEUED = 1
|
|
ACTIVE = 2
|
|
|
|
|
|
@dataclass
|
|
class QueueWorkerInfo:
|
|
app_name: str
|
|
app_title: str
|
|
worker_type: str
|
|
state: QueueWorkerState
|
|
progress: int = 0
|
|
|
|
|
|
class QueueWorker(Worker):
|
|
"""
|
|
Base queueable worker class
|
|
|
|
This class is a specialization of the `Worker` class. It provides feedback signals to know
|
|
if a worker has started or finished.
|
|
|
|
To use this class you have to assign the signals object of your concrete implementation
|
|
to the `QueueWorker.signals` attribute, implement `QueueWorker.run_real()` and `QueueWorker.worker_info()`
|
|
"""
|
|
|
|
class Signals(QObject):
|
|
started = pyqtSignal()
|
|
finished = pyqtSignal()
|
|
|
|
def __init__(self):
|
|
super(QueueWorker, self).__init__()
|
|
self.feedback = QueueWorker.Signals()
|
|
self.state = QueueWorkerState.QUEUED
|
|
self._kill = False
|
|
|
|
@pyqtSlot()
|
|
def run(self):
|
|
self.state = QueueWorkerState.ACTIVE
|
|
self.feedback.started.emit()
|
|
super(QueueWorker, self).run()
|
|
self.feedback.finished.emit()
|
|
self.feedback.deleteLater()
|
|
|
|
@abstractmethod
|
|
def worker_info(self) -> QueueWorkerInfo:
|
|
pass
|
|
|
|
def kill(self):
|
|
raise NotImplementedError
|
|
self._kill = True
|
|
|