Wobuzz/wobuzz/ui/process/process_dock.py

#!/usr/bin/python3

from PyQt6.QtCore import QTimer, pyqtSignal
from PyQt6.QtWidgets import QWidget, QDockWidget, QScrollArea, QVBoxLayout

from .process import BackgroundProcess

PROGRESS_UPDATE_RATE = 10
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, str, int, object)

    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_title: str, description: str, steps: int, getter):
        process = BackgroundProcess(
            job_title,
            self.process_container,
            description,
            steps
        )

        if getter is not None:
            process.get_progress = getter

        self.add_process(job_title, process)

    def on_background_job_stop(self, job):
        if job in self.processes:
            self.processes.pop(job).deleteLater()