72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
#!/usr/bin/python3
|
|
|
|
from PyQt6.QtCore import QTimer, pyqtSignal
|
|
from PyQt6.QtWidgets import QWidget, QDockWidget, QScrollArea, QVBoxLayout
|
|
|
|
from .process import BackgroundProcess
|
|
|
|
PROGRESS_UPDATE_RATE = 30
|
|
PROGRESS_UPDATE_INTERVAL = 1000 // PROGRESS_UPDATE_RATE
|
|
|
|
|
|
class ProcessDock(QDockWidget):
|
|
# we need a signal for self.on_background_job_start() because PyQt6 doesn't allow some operations to be performed
|
|
# from a different thread
|
|
job_started_signal = pyqtSignal(str)
|
|
|
|
def __init__(self, app, parent=None):
|
|
super().__init__(parent)
|
|
|
|
self.app = app
|
|
|
|
self.processes = {}
|
|
|
|
self.setWindowTitle("Background Processes")
|
|
|
|
self.scroll_area = QScrollArea(self)
|
|
self.scroll_area.setWidgetResizable(True)
|
|
|
|
self.process_container = QWidget(self.scroll_area)
|
|
|
|
self.process_layout = QVBoxLayout(self.process_container)
|
|
|
|
# add expanding widget so the distance between processes will be equal
|
|
self.process_layout.addWidget(QWidget(self.process_container))
|
|
|
|
self.process_container.setLayout(self.process_layout)
|
|
|
|
self.scroll_area.setWidget(self.process_container)
|
|
|
|
self.setWidget(self.scroll_area)
|
|
|
|
self.progress_update_timer = QTimer()
|
|
self.progress_update_timer.timeout.connect(self.update_processes)
|
|
self.progress_update_timer.start(PROGRESS_UPDATE_INTERVAL)
|
|
|
|
self.job_started_signal.connect(self.on_background_job_start)
|
|
|
|
def add_process(self, name: str, process: BackgroundProcess):
|
|
if not name in self.processes:
|
|
self.processes[name] = process
|
|
self.process_layout.insertWidget(self.process_layout.count() - 1, process)
|
|
|
|
def update_processes(self):
|
|
for process in self.processes.values():
|
|
process.update_progress()
|
|
|
|
def on_background_job_start(self, job):
|
|
match job:
|
|
case "track_caching":
|
|
self.add_process(
|
|
job,
|
|
BackgroundProcess(
|
|
"Loading Track",
|
|
self.process_container,
|
|
"Loading next track in the background so it starts immediately."
|
|
)
|
|
)
|
|
|
|
def on_background_job_stop(self, job):
|
|
if job in self.processes:
|
|
self.processes.pop(job).deleteLater()
|
|
|