如何在用 pickle 加载或转储时阻止动画 QCursor 冻结?

How to stop an animated QCursor from freezing when loading or dumping with pickle?

作为 post 的后续问题,我想知道是否可以扩展游标的功能,以便在使用 pickle 转储或保存数据时,光标的动画不会冻结。

from PyQt5 import QtCore, QtGui, QtWidgets
import pickle
import gzip
import numpy as np


class ManagerCursor(QtCore.QObject):
    def __init__(self, parent=None):
        super(ManagerCursor, self).__init__(parent)
        self._movie = None
        self._widget = None
        self._last_cursor = None

    def setMovie(self, movie):
        if isinstance(self._movie, QtGui.QMovie):
            if not self._movie != QtGui.QMovie.NotRunning:
                self._movie.stop()
            del self._movie
        self._movie = movie
        self._movie.frameChanged.connect(self.on_frameChanged)
        self._movie.started.connect(self.on_started)
        self._movie.finished.connect(self.restore_cursor)

    def setWidget(self, widget):
        self._widget = widget

    @QtCore.pyqtSlot()
    def on_started(self):
        if self._widget is not None:
            self._last_cursor = self._widget.cursor()

    @QtCore.pyqtSlot()
    def restore_cursor(self):
        if self._widget is not None:
            if self._last_cursor is not None:
                self._widget.setCursor(self._last_cursor)
        self._last_cursor = None

    @QtCore.pyqtSlot()
    def start(self):
        if self._movie is not None:
            self._movie.start()

    @QtCore.pyqtSlot()
    def stop(self):
        if self._movie is not None:
            self._movie.stop()
            self.restore_cursor()

    @QtCore.pyqtSlot()
    def on_frameChanged(self):
        pixmap = self._movie.currentPixmap()
        cursor = QtGui.QCursor(pixmap)
        if self._widget is not None:
            if self._last_cursor is None:
                self._last_cursor = self._widget.cursor()
            self._widget.setCursor(cursor)


class Progress(QtWidgets.QDialog):
    def __init__(self):
        super().__init__()
        progress = QtWidgets.QProgressBar()
        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(progress)


class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super(Widget, self).__init__(parent)
        start_btn = QtWidgets.QPushButton("start", clicked=self.on_start)
        stop_btn = QtWidgets.QPushButton("stop", clicked=self.on_stop)
        dump_btn = QtWidgets.QPushButton("dump", clicked=self.dump)
        load_btn = QtWidgets.QPushButton("load", clicked=self.load)

        self.file = 'test'

        text_edit = QtWidgets.QTextEdit()
        self.popup = None

        self._manager = ManagerCursor(self)
        movie = QtGui.QMovie('../assets/comet_resized.gif')
        self._manager.setMovie(movie)
        self._manager.setWidget(self)

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(start_btn)
        lay.addWidget(stop_btn)
        lay.addWidget(dump_btn)
        lay.addWidget(load_btn)
        lay.addWidget(text_edit)
        lay.addStretch()

    @QtCore.pyqtSlot()
    def dump(self):
        self._manager.start()
        self.popup = Progress()
        self.popup.show()
        data = [np.full(1000, 1000) for i in range(100000)]
        with gzip.open(self.file, 'wb') as output_file:
            pickle.dump(data, output_file, pickle.HIGHEST_PROTOCOL)

    def load(self):
        self._manager.start()
        self.popup = Progress()
        self.popup.show()
        with gzip.open(self.file, 'rb') as input_file:
            data = pickle.load(input_file)

    @QtCore.pyqtSlot()
    def on_start(self):
        self._manager.start()

    @QtCore.pyqtSlot()
    def on_stop(self):
        self._manager.stop()


if __name__ == '__main__':
    import sys
    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.resize(640, 480)
    w.show()
    sys.exit(app.exec_())

不应在主线程中执行繁重的任务,因为它们会通过冻结 GUI 来阻塞事件循环,必须在另一个线程中执行。

from functools import partial
import gzip
import pickle

from PyQt5 import QtCore, QtGui, QtWidgets

import numpy as np


class Worker(QtCore.QObject):
    dumpStarted = QtCore.pyqtSignal()
    dumpFinished = QtCore.pyqtSignal()

    dataChanged = QtCore.pyqtSignal(object)
    loadStarted = QtCore.pyqtSignal()
    loadFinished = QtCore.pyqtSignal()

    @QtCore.pyqtSlot(str, object)
    def dump(self, filename, data):
        self.dumpStarted.emit()
        with gzip.open(filename, "wb") as output_file:
            pickle.dump(data, output_file, pickle.HIGHEST_PROTOCOL)
        self.dumpFinished.emit()

    @QtCore.pyqtSlot(str)
    def load(self, filename):
        self.loadStarted.emit()
        with gzip.open(filename, "rb") as input_file:
            data = pickle.load(input_file)
            self.dataChanged.emit(data)
        self.loadFinished.emit()


class ManagerCursor(QtCore.QObject):
    def __init__(self, parent=None):
        super(ManagerCursor, self).__init__(parent)
        self._movie = None
        self._widget = None
        self._last_cursor = None

    def setMovie(self, movie):
        if isinstance(self._movie, QtGui.QMovie):
            if not self._movie != QtGui.QMovie.NotRunning:
                self._movie.stop()
            del self._movie
        self._movie = movie
        self._movie.frameChanged.connect(self.on_frameChanged)
        self._movie.started.connect(self.on_started)
        self._movie.finished.connect(self.restore_cursor)

    def setWidget(self, widget):
        self._widget = widget

    @QtCore.pyqtSlot()
    def on_started(self):
        if self._widget is not None:
            self._last_cursor = self._widget.cursor()

    @QtCore.pyqtSlot()
    def restore_cursor(self):
        if self._widget is not None:
            if self._last_cursor is not None:
                self._widget.setCursor(self._last_cursor)
        self._last_cursor = None

    @QtCore.pyqtSlot()
    def start(self):
        if self._movie is not None:
            self._movie.start()

    @QtCore.pyqtSlot()
    def stop(self):
        if self._movie is not None:
            self._movie.stop()
            self.restore_cursor()

    @QtCore.pyqtSlot()
    def on_frameChanged(self):
        pixmap = self._movie.currentPixmap()
        cursor = QtGui.QCursor(pixmap)
        if self._widget is not None:
            if self._last_cursor is None:
                self._last_cursor = self._widget.cursor()
            self._widget.setCursor(cursor)


class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super(Widget, self).__init__(parent)
        start_btn = QtWidgets.QPushButton("start", clicked=self.on_start)
        stop_btn = QtWidgets.QPushButton("stop", clicked=self.on_stop)
        dump_btn = QtWidgets.QPushButton("dump", clicked=self.dump)
        load_btn = QtWidgets.QPushButton("load", clicked=self.load)

        self.file = "test"

        text_edit = QtWidgets.QTextEdit()
        self.popup = None

        self._manager = ManagerCursor(self)
        movie = QtGui.QMovie("giphy.gif")
        self._manager.setMovie(movie)
        self._manager.setWidget(self)

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(start_btn)
        lay.addWidget(stop_btn)
        lay.addWidget(dump_btn)
        lay.addWidget(load_btn)
        lay.addWidget(text_edit)
        lay.addStretch()

        thread = QtCore.QThread(self)
        thread.start()

        self._worker = Worker()
        self._worker.moveToThread(thread)

        self._worker.dumpStarted.connect(self._manager.start)
        self._worker.dumpFinished.connect(self._manager.stop)
        self._worker.dumpStarted.connect(partial(dump_btn.setEnabled, False))
        self._worker.dumpFinished.connect(partial(dump_btn.setEnabled, True))

        self._worker.loadStarted.connect(self._manager.start)
        self._worker.loadFinished.connect(self._manager.stop)
        self._worker.loadStarted.connect(partial(load_btn.setEnabled, False))
        self._worker.loadFinished.connect(partial(load_btn.setEnabled, True))
        self._worker.dataChanged.connect(self.on_data_changed)

    @QtCore.pyqtSlot()
    def dump(self):
        data = [np.full(1000, 1000) for i in range(100000)]
        wrapper = partial(self._worker.dump, self.file, data)
        QtCore.QTimer.singleShot(0, wrapper)

    @QtCore.pyqtSlot()
    def load(self):
        wrapper = partial(self._worker.load, self.file)
        QtCore.QTimer.singleShot(0, wrapper)

    @QtCore.pyqtSlot(object)
    def on_data_changed(self, data):
        print(data)

    @QtCore.pyqtSlot()
    def on_start(self):
        self._manager.start()

    @QtCore.pyqtSlot()
    def on_stop(self):
        self._manager.stop()


if __name__ == "__main__":
    import sys

    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.resize(640, 480)
    w.show()
    sys.exit(app.exec_())