来自长进程的 PySide6 并发信息窗口（有延迟）

Question

我正在做一个小型 PySide6 项目，我需要在其中运行从轻到重的计算。 由于计算的持续时间（0.1 - 600 秒），我想在计算时间超过 3 秒时显示一个小窗口，指示“正在处理......”

我尝试创建一个单独的 QThread，在其中我从运行计算的主窗口/线程接收信号以在延迟一段时间后触发响应。

这是我的示例代码：

from PySide6.QtWidgets import QWidget, QPushButton, QVBoxLayout, QApplication, QLabel
from PySide6.QtCore import Slot, Signal, QObject, QThread

import time
import sys


class AnotherWindow(QWidget):
    def __init__(self):
        super().__init__()
        layout = QVBoxLayout()
        self.label = QLabel("Processing ...")
        layout.addWidget(self.label)
        self.setLayout(layout)


class MyWorker(QObject):
    display = Signal()

    @Slot()
    def display_trigger(self):
        print('before delay')
        time.sleep(2)
        print('delay done')
        self.display.emit()


class Window(QWidget):
    def __init__(self):
        super(Window, self).__init__()

        self.thread = None
        self.worker = None
        self.button = None
        self.w = AnotherWindow()

        self.init_ui()
        self.setup_thread()

    def init_ui(self):
        layout = QVBoxLayout()
        self.button = QPushButton('User input')
        self.button.setEnabled(True)
        layout.addWidget(self.button)
        self.setLayout(layout)
        self.show()

    @Slot()
    def display(self):
        print("display it !")
        self.w.show()

    @Slot()
    def processing(self):
        print("=========")
        for i in range(8):
            print(i)
            time.sleep(1)
        self.w = None

    def closeEvent(self, event):
        # self.w = None
        self.thread.quit()
        del self.thread

    def setup_thread(self):
        self.thread = QThread()
        self.worker = MyWorker()

        self.worker.moveToThread(self.thread)

        self.worker.display.connect(self.display)
        self.button.clicked.connect(self.worker.display_trigger)
        self.button.clicked.connect(self.processing)

        self.thread.start()


if __name__ == "__main__":
    app = QApplication(sys.argv)
    w = Window()
    w.show()
    sys.exit(app.exec())

这里的问题是在计算的过程中没有出现‘processing’窗口，然后明显报错，因为在processing方法最后删除了它却试图显示。我想保留这个删除，因为一旦处理完成，窗口就没用了，但我不明白为什么当信号被触发时小部件不显示。似乎在长进程运行时信号被阻塞了。关于主线程中的这个阻塞信号有什么想法吗？

Answer 1

经过一番调优，找到了解决方案，感谢“著名在线AI”和前人的建议。 我做错了：我颠倒了问题和线程。在下一个解决方案中，弹出窗口只是从主线程启动，而长计算则从另一个线程启动。这是做这件事的好方法。

这是我最终调整后的工作代码：

import sys
import time
from PySide6 import QtWidgets, QtCore, QtGui


class ProcessingWindow(QtWidgets.QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setModal(True)
        self.setWindowTitle("Processing...")
        self.setFixedSize(200, 100)
        layout = QtWidgets.QVBoxLayout(self)
        self.label = QtWidgets.QLabel("Processing...", self)
        self.label.setFont(QtGui.QFont("Arial", 16))
        layout.addWidget(self.label, alignment=QtCore.Qt.AlignmentFlag.AlignCenter)
        self.setWindowFlags(QtCore.Qt.WindowType.SplashScreen)

    def center(self):
        # Calculate the center position of the main window
        parent_rect = self.parent().geometry()
        parent_center = parent_rect.center()

        # Calculate the position of the dialog
        dialog_rect = self.geometry()
        dialog_rect.moveCenter(parent_center)
        self.setGeometry(dialog_rect)


class LongProcessThread(QtCore.QThread):
    finished = QtCore.Signal()

    def __init__(self, parent=None, duration=None):
        super().__init__(parent)
        self.parent = parent
        self.duration = duration

    def run(self):
        self.parent.simulate_long_process(self.duration)
        self.finished.emit()


class MainWindow(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Long Process Example")
        self.setFixedSize(500, 300)
        layout = QtWidgets.QVBoxLayout(self)
        self.button = QtWidgets.QPushButton("Run Long Process", self)
        layout.addWidget(self.button, alignment=QtCore.Qt.AlignmentFlag.AlignCenter)
        self.button.clicked.connect(self.run_long_process)
        self.processing_window = None

    def run_long_process(self):
        self.button.setEnabled(False)
        long_process_duration = 10  # seconds
        delay_popup = 3000  # milliseconds
        long_process_thread = LongProcessThread(self, long_process_duration)
        long_process_thread.finished.connect(self.enable_button)
        long_process_thread.start()
        QtCore.QTimer.singleShot(delay_popup, lambda: self.show_processing_window(long_process_thread))

    @staticmethod
    def simulate_long_process(duration=None):
        print("before")
        if duration:
            time.sleep(duration)  # Simulating a long process
        print(f"after {duration} seconds")

    def show_processing_window(self, long_process_thread):
        if not long_process_thread.isFinished():
            self.processing_window = ProcessingWindow(self)
            self.processing_window.center()
            self.processing_window.show()

    def enable_button(self):
        self.processing_window.close()
        self.button.setEnabled(True)


if __name__ == "__main__":
    app = QtWidgets.QApplication(sys.argv)
    main_window = MainWindow()
    main_window.show()
    sys.exit(app.exec())