How do I queue QProcesses in PyQt5?

Question

I want to queue QProcess in PyQt5 or simply block while still reading the stdout with readAll(). The equivalent of subprocess.call instead of subprocess.Pop. When using waitForFinished() the stdout with readAll() will all come at once when the process is ended instead of flowing out while it's processing.

Example script:

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()

Answer 1

The solution in this case is to create a TaskManager class that is responsible for handling the sequentiality between the tasks.

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:

Generalizing the problem, it can be said that the n-th process needs default arguments and additional arguments.

• The default arguments are independent and fixed

• The additional arguments depend on the previous process through a certain function.

So it can be generalized using the following expressions:

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

or using the following diagram:

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

Then to structure the process the following dictionary is created:

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

Where fun is the function used to process the output of this task to obtain additional arguments for the next task.

In the following example, I will use scriptX.py as a program instead of 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 is the function that uses the result of process-1 to generate the additional argument required by process-2 and must return it. similar case for fun2to3

def fun1to2(*args):
    return "additional_arg_for_process2_from_result1" 

def fun2to3(*args):
    return "additional_arg_for_process3_from_result2" 

so based on the above we create the tasks:

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"]}]

Using all of the above, the final implementation is as follows:

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: