如何在 PyQt5 中对 QProcess 进行排队？

我想在 PyQt5 中对 QProcess 进行排队，或者只是阻塞，同时仍然使用 readAll() 读取标准输出。相当于 subprocess.call 而不是 subprocess.Pop。当使用 waitForFinished() 时，带有 readAll() 的标准输出将在进程结束时立即全部出现，而不是在处理时流出。

示例脚本：

from PIL import Image
import numpy as np
import sys
from PyQt5 import QtGui,QtCore, QtWidgets

class gui(QtWidgets.QMainWindow):
    def __init__(self):
        super(gui, self).__init__()
        self.initUI()

    def dataReady(self):
        cursor = self.output.textCursor()
        cursor.movePosition(cursor.End)
        cursor.insertText(str(self.process.readAll(), "utf-8"))
        self.output.ensureCursorVisible()


    def callProgram(self):
        # run the process
        # `start` takes the exec and a list of argument

        filepath = 'path\image.tif'

        self.process.start('some_command filepath'])

        # This will output a file image.tif specified by filepath: 

        # Import file and do stuff to it:

        # E.g.

        im = Image.open('filepath')

        imarray = np.array(im)

        # Get image extents as argument to next process:

        ext = str(imarray.size)


        imarray[imarray == 10] = 5

        # Save changes

        im = Image.fromarray(imarray)
        im.save(filepath)            

        # Now the image has been updated and should be in a new process below

        cmd = 'some_other_command' + filepath + ext

        self.process.start(cmd)

        # Same thing goes on here:

        self.process.start('some_command filepath')

        # Import file once again

        im = Image.open('filepath')

        imarray[imarray == 10] = 5

        # Save changes

        im = Image.fromarray(imarray)
        im.save(filepath)    

    def initUI(self):

        layout = QtWidgets.QHBoxLayout()
        self.runButton = QtWidgets.QPushButton('Run')
        self.runButton.clicked.connect(self.callProgram)

        self.output = QtWidgets.QTextEdit()

        layout.addWidget(self.output)
        layout.addWidget(self.runButton)

        centralWidget = QtWidgets.QWidget()
        centralWidget.setLayout(layout)
        self.setCentralWidget(centralWidget)

        # QProcess object for external app
        self.process = QtCore.QProcess(self)
        # QProcess emits `readyRead` when there is data to be read
        self.process.readyRead.connect(self.dataReady)

        # Just to prevent accidentally running multiple times
        # Disable the button when process starts, and enable it when it finishes
        self.process.started.connect(lambda: self.runButton.setEnabled(False))
        self.process.finished.connect(lambda: self.runButton.setEnabled(True))


#Function Main Start
def main():
    app = QtWidgets.QApplication(sys.argv)
    ui=gui()
    ui.show()
    sys.exit(app.exec_())
#Function Main END

if __name__ == '__main__':
    main()

这种情况下的解决方案是创建一个TaskManager负责处理任务之间的顺序的类。

import sys

from PyQt5 import QtCore, QtWidgets
from functools import partial

class TaskManager(QtCore.QObject):
    started = QtCore.pyqtSignal()
    finished = QtCore.pyqtSignal()
    progressChanged = QtCore.pyqtSignal(int, QtCore.QByteArray)

    def __init__(self, parent=None):
        QtCore.QObject.__init__(self, parent)
        self._process = QtCore.QProcess(self)
        self._process.finished.connect(self.handleFinished)
        self._progress = 0

    def start_tasks(self, tasks):
        self._tasks = iter(tasks)
        self.fetchNext()
        self.started.emit()
        self._progress = 0

    def fetchNext(self):
            try:
                task = next(self._tasks)
            except StopIteration:
                return False
            else:
                self._process.start(*task)
            return True

    def processCurrentTask(self):
        output = self._process.readAllStandardOutput()
        self._progress += 1
        self.progressChanged.emit(self._progress, output)

    def handleFinished(self):
        self.processCurrentTask()
        if not self.fetchNext():
            self.finished.emit()



class gui(QtWidgets.QMainWindow):
    def __init__(self):
        super(gui, self).__init__()
        self.initUI()


    def dataReady(self, progress, result):
        self.output.append(str(result, "utf-8"))
        self.progressBar.setValue(progress)

    def callProgram(self):
        tasks = [("ping", ["8.8.8.8"]),
                 ("ping", ["8.8.8.8"]),
                 ("ping", ["8.8.8.8"])]

        self.progressBar.setMaximum(len(tasks))
        self.manager.start_tasks(tasks)

    def initUI(self):
        layout = QtWidgets.QVBoxLayout()
        self.runButton = QtWidgets.QPushButton('Run')

        self.runButton.clicked.connect(self.callProgram)

        self.output = QtWidgets.QTextEdit()

        self.progressBar = QtWidgets.QProgressBar()

        layout.addWidget(self.output)
        layout.addWidget(self.runButton)
        layout.addWidget(self.progressBar)

        centralWidget = QtWidgets.QWidget()
        centralWidget.setLayout(layout)
        self.setCentralWidget(centralWidget)

        self.manager = TaskManager(self)
        self.manager.progressChanged.connect(self.dataReady)
        self.manager.started.connect(partial(self.runButton.setEnabled, False))
        self.manager.finished.connect(partial(self.runButton.setEnabled, True))


def main():
    app = QtWidgets.QApplication(sys.argv)
    ui=gui()
    ui.show()
    sys.exit(app.exec_())

if __name__ == '__main__':
    main()

Update:

概括这个问题，可以说第 n 个进程需要默认参数和附加参数。

• 默认参数是独立且固定的

• 附加参数取决于之前通过某个函数进行的过程。

因此可以使用以下表达式进行概括：

result_n = process_n(default_arguments, additional_args_n)`
additional_args_n = fun_n(result_n-1)`

或使用下图：

 ________      _________      ________      _________      ________
|        |    |         |    |        |    |         |    |        |
|        |    |         |    |        |    |         |    |        |
| TASK-1 |--->| FUN1TO2 |--->| TASK-2 |--->| FUN2TO3 |--->| TASK-3 |
|        |    |         |    |        |    |         |    |        |
|________|    |_________|    |________|    |_________|    |________|

然后为了构建该过程，创建以下字典：

task_n = {"program": program, "args": default_arguments, "function": fun}

Where fun是用于处理此任务的输出以获得下一个任务的附加参数的函数。

在下面的例子中，我将使用scriptX.py作为一个程序而不是ping.

#script1.py
import sys

def foo(*args):
    v,  = args
    return "1-"+"".join(v)

arg = sys.argv[1:]
print(foo(arg))

#script2.py
import sys

def foo(*args):
    v,  = args
    return "2-"+"".join(v)

arg = sys.argv[1:]
print(foo(arg))

#script3.py
import sys

def foo(*args):
    v,  = args
    return "3-"+"".join(v)

arg = sys.argv[1:]
print(foo(arg))

fun1to2是使用 process-1 的结果生成 process-2 所需的附加参数并且必须返回它的函数。类似的情况fun2to3

def fun1to2(*args):
    return "additional_arg_for_process2_from_result1" 

def fun2to3(*args):
    return "additional_arg_for_process3_from_result2" 

因此，基于上述内容，我们创建任务：

tasks = [{"program": "python", "args": ["scripts/script1.py", "default_argument1"], "function": fun1to2},
         {"program": "python", "args": ["scripts/script2.py", "default_argument2"], "function": fun2to3},
         {"program": "python", "args": ["scripts/script3.py", "default_argument3"]}]

使用以上所有内容，最终实现如下：

import sys

from PyQt5 import QtCore, QtWidgets
from functools import partial

class TaskManager(QtCore.QObject):
    started = QtCore.pyqtSignal()
    finished = QtCore.pyqtSignal()
    progressChanged = QtCore.pyqtSignal(int, QtCore.QByteArray)

    def __init__(self, parent=None):
        QtCore.QObject.__init__(self, parent)
        self._process = QtCore.QProcess(self)
        self._process.finished.connect(self.handleFinished)
        self._progress = 0
        self._currentTask = None

    def start_tasks(self, tasks):
        self._tasks = iter(tasks)
        self.fetchNext()
        self.started.emit()
        self._progress = 0

    def fetchNext(self, additional_args=None):
            try:
                self._currentTask = next(self._tasks)
            except StopIteration:
                return False
            else:
                program = self._currentTask.get("program")
                args = self._currentTask.get("args")
                if additional_args is not None:
                    args += additional_args
                self._process.start(program, args)
            return True

    def processCurrentTask(self):
        output = self._process.readAllStandardOutput()
        self._progress += 1
        fun = self._currentTask.get("function")
        res = None
        if fun:
            res = fun(output)
        self.progressChanged.emit(self._progress, output)
        return res

    def handleFinished(self):
        args = self.processCurrentTask()
        if not self.fetchNext(args):
            self.finished.emit()


def fun1to2(args):
    return "-additional_arg_for_process2_from_result1" 

def fun2to3(args):
    return "-additional_arg_for_process3_from_result2" 

class gui(QtWidgets.QMainWindow):
    def __init__(self):
        super(gui, self).__init__()
        self.initUI()


    def dataReady(self, progress, result):
        self.output.append(str(result, "utf-8"))
        self.progressBar.setValue(progress)


    def callProgram(self):
        tasks = [{"program": "python", "args": ["scripts/script1.py", "default_argument1"], "function": fun1to2},
                 {"program": "python", "args": ["scripts/script2.py", "default_argument2"], "function": fun2to3},
                 {"program": "python", "args": ["scripts/script3.py", "default_argument3"]}]

        self.progressBar.setMaximum(len(tasks))
        self.manager.start_tasks(tasks)

    def initUI(self):
        layout = QtWidgets.QVBoxLayout()
        self.runButton = QtWidgets.QPushButton('Run')

        self.runButton.clicked.connect(self.callProgram)

        self.output = QtWidgets.QTextEdit()

        self.progressBar = QtWidgets.QProgressBar()

        layout.addWidget(self.output)
        layout.addWidget(self.runButton)
        layout.addWidget(self.progressBar)

        centralWidget = QtWidgets.QWidget()
        centralWidget.setLayout(layout)
        self.setCentralWidget(centralWidget)

        self.manager = TaskManager(self)
        self.manager.progressChanged.connect(self.dataReady)
        self.manager.started.connect(partial(self.runButton.setEnabled, False))
        self.manager.finished.connect(partial(self.runButton.setEnabled, True))


def main():
    app = QtWidgets.QApplication(sys.argv)
    ui=gui()
    ui.show()
    sys.exit(app.exec_())

if __name__ == '__main__':
    main()

Result: