Welcome to OGeek Q&A Community for programmer and developer-Open, Learning and Share
Welcome To Ask or Share your Answers For Others

Categories

0 votes
511 views
in Technique[技术] by (71.8m points)

python - Run command with PyQt5 and getting the stdout and stderr

I want to run command with PyQt5. And I want to get the stdout and stderr in time order and in real-time.

I separated into UI class and Worker class. There are several UI classes, but for simplicity, I've specified just one.

I've tried to solve this problem, but I can't. I can't connect between the Worker thread and logger function.

test_ui.py

import sys
import subprocess
from PyQt5.QtWidgets import QApplication, QWidget, QHBoxLayout, QVBoxLayout
from PyQt5.QtWidgets import QPushButton, QTextEdit
from worker import Worker


class TestUI(QWidget):
    def __init__(self):
        super().__init__()
        self.worker = Worker()
        self.btn1 = QPushButton("Button1")
        self.btn2 = QPushButton("Button2")
        self.btn3 = QPushButton("Button3")
        self.result = QTextEdit()
        self.init_ui()

    def init_ui(self):
        self.btn1.clicked.connect(self.press_btn1)
        self.btn2.clicked.connect(self.press_btn2)
        self.btn3.clicked.connect(self.press_btn3)

        hlayout1 = QHBoxLayout()
        hlayout1.addWidget(self.btn1)
        hlayout1.addWidget(self.btn2)
        hlayout1.addWidget(self.btn3)

        hlayout2 = QHBoxLayout()
        hlayout2.addWidget(self.result)

        vlayout = QVBoxLayout()
        vlayout.addLayout(hlayout1)
        vlayout.addLayout(hlayout2)

        self.setLayout(vlayout)
        self.show()

    def press_btn1(self):
        command1 = "dir"
        path = "./"
        self.worker.run_command(command1, path)
        self.worker.outSignal.connect(self.logging)

    def press_btn2(self):
        command2 = "cd"
        path = "./"
        self.worker.run_command(command2, path)
        self.worker.outSignal.connect(self.logging)

    def press_btn3(self):
        command3 = "whoami"
        path = "./"
        self.worker.run_command(command3, path)
        self.worker.outSignal.connect(self.logging)

    def logging(self, str):
        self.result.append(str.strip())


if __name__ == "__main__":
    APP = QApplication(sys.argv)
    ex = TestUI()
    sys.exit(APP.exec_())

worker.py


from PyQt5.QtCore import QProcess, pyqtSignal


class Worker:
    outSignal = pyqtSignal(str)
    errSignal = pyqtSignal(str)

    def __init__(self):
        self.proc = QProcess()

    def run_command(self, cmd, path):
        self.proc.setWorkingDirectory(path)
        self.proc.setProcessChannelMode(QProcess.MergedChannels)
        self.proc.readyReadStandardOutput.connect(self.onReadyStandardOutput)
        self.proc.finished.connect(self.proc.deleteLater)
        self.proc.start(cmd)

    def onReadyStandardOutput(self):
        result = self.proc.readAllStandardOutput().data().decode()
        self.outSignal.emit(result)

    def onReadyStandardError(self):
        result = self.proc.readAllStandardError().data().decode()
        self.errSignal.emit(result)

Update:

Applying here solution and making the following modifications still fails the code:

@pyqtSlot()
def press_btn1(self):
    command1 = "dir"
    path = "./"
    self.worker.run_command(command1, path)

@pyqtSlot()
def press_btn2(self):
    command2 = "cd"
    path = "./"
    self.worker.run_command(command2, path)

@pyqtSlot()
def press_btn3(self):
    command3 = "test.bat"
    path = "./"
    self.worker.run_command(command3, path)

@pyqtSlot(str)
def logging(self, msg):
    msg = msg.strip()
    if msg != "":
        self.result.append(msg)

test.bat

@echo off

echo "Output 1"
timeout /t 1
1>&2 echo "Error 1"
timeout /t 1
echo "Output 2"
timeout /t 1
1>&2 echo "Error 2"

Batchfile Issue

This is the result when I run it through the command prompt.

It outputs one line every second in real-time.

"Output 1"

Waiting for 0 seconds, press a key to continue ...
"Error 1"

Waiting for 0 seconds, press a key to continue ...
"Output 2"

Waiting for 0 seconds, press a key to continue ...
"Error 2"

This is the result of the application.

It outputs whole lines after 3 seconds. And the time order is not right.

"Output 1"

Waiting for 1 seconds, press a key to continue ...0

Waiting for 1 seconds, press a key to continue ...0
"Output 2"

Waiting for 1 seconds, press a key to continue ...0
"Error 1"
"Error 2"
See Question&Answers more detail:os

与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…
Welcome To Ask or Share your Answers For Others

1 Reply

0 votes
by (71.8m points)

You have the following errors:

  • The signals only work in the QObjects so it is necessary for Worker to inherit from QObject.

  • It is recommended that QProcess not be a member of the class since we say that task 1 is being executed and without finishing you try to execute task 2 so that task 1 will be replaced which is not what you want, instead QProcess can be done be a child of Worker so that your life cycle is not limited to the method where it was created.

  • If you want to monitor the stderr and stdio output separately then you should not like processChannelMode to QProcess::MergedChannels since this will join both outputs, on the other hand if the above is eliminated then you must use the readyReadStandardError signal to know when stderr is modified.

  • Since QProcess is not a member of the class, it is difficult to obtain the QProcess in onReadyStandardOutput and onReadyStandardError, but for this you must use the sender() method that has the object that emitted the signal.

  • The connections between signals and slot should only be made once, in your case you do it in press_btn1, press_btn2 and press_btn3 so you will get 3 times the same information.

  • Do not use str since it is a built-in function.

Considering the above, the solution is:

worker.py

from PyQt5.QtCore import QObject, QProcess, pyqtSignal, pyqtSlot


class Worker(QObject):
    outSignal = pyqtSignal(str)
    errSignal = pyqtSignal(str)

    def run_command(self, cmd, path):
        proc = QProcess(self)
        proc.setWorkingDirectory(path)
        proc.readyReadStandardOutput.connect(self.onReadyStandardOutput)
        proc.readyReadStandardError.connect(self.onReadyStandardError)
        proc.finished.connect(proc.deleteLater)
        proc.start(cmd)

    @pyqtSlot()
    def onReadyStandardOutput(self):
        proc = self.sender()
        result = proc.readAllStandardOutput().data().decode()
        self.outSignal.emit(result)

    @pyqtSlot()
    def onReadyStandardError(self):
        proc = self.sender()
        result = proc.readAllStandardError().data().decode()
        self.errSignal.emit(result)

test_ui.py

import sys

from PyQt5.QtCore import pyqtSlot
from PyQt5.QtWidgets import QApplication, QGridLayout, QPushButton, QTextEdit, QWidget

from worker import Worker


class TestUI(QWidget):
    def __init__(self):
        super().__init__()
        self.worker = Worker()
        self.worker.outSignal.connect(self.logging)
        self.btn1 = QPushButton("Button1")
        self.btn2 = QPushButton("Button2")
        self.btn3 = QPushButton("Button3")
        self.result = QTextEdit()
        self.init_ui()

    def init_ui(self):
        self.btn1.clicked.connect(self.press_btn1)
        self.btn2.clicked.connect(self.press_btn2)
        self.btn3.clicked.connect(self.press_btn3)

        lay = QGridLayout(self)
        lay.addWidget(self.btn1, 0, 0)
        lay.addWidget(self.btn2, 0, 1)
        lay.addWidget(self.btn3, 0, 2)
        lay.addWidget(self.result, 1, 0, 1, 3)

    @pyqtSlot()
    def press_btn1(self):
        command1 = "dir"
        path = "./"
        self.worker.run_command(command1, path)

    @pyqtSlot()
    def press_btn2(self):
        command2 = "cd"
        path = "./"
        self.worker.run_command(command2, path)

    @pyqtSlot()
    def press_btn3(self):
        command3 = "whoami"
        path = "./"
        self.worker.run_command(command3, path)

    @pyqtSlot(str)
    def logging(self, string):
        self.result.append(string.strip())


if __name__ == "__main__":
    APP = QApplication(sys.argv)
    ex = TestUI()
    ex.show()
    sys.exit(APP.exec_())

Update:

QProcess has limitations to execute console commands such as .bat so in this case you can use subprocess.Popen by executing it in another thread and sending the information through signals:

worker.py

import subprocess
import threading

from PyQt5 import QtCore


class Worker(QtCore.QObject):
    outSignal = QtCore.pyqtSignal(str)

    def run_command(self, cmd, **kwargs):
        threading.Thread(
            target=self._execute_command, args=(cmd,), kwargs=kwargs, daemon=True
        ).start()

    def _execute_command(self, cmd, **kwargs):
        proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kwargs
        )
        for line in proc.stdout:
            self.outSignal.emit(line.decode())

test_ui.py

import sys

from PyQt5.QtCore import pyqtSlot
from PyQt5.QtWidgets import QApplication, QGridLayout, QPushButton, QTextEdit, QWidget

from worker import Worker


class TestUI(QWidget):
    def __init__(self):
        super().__init__()
        self.worker = Worker()
        self.worker.outSignal.connect(self.logging)
        self.btn1 = QPushButton("Button1")
        self.btn2 = QPushButton("Button2")
        self.btn3 = QPushButton("Button3")
        self.result = QTextEdit()
        self.init_ui()

    def init_ui(self):
        self.btn1.clicked.connect(self.press_btn1)
        self.btn2.clicked.connect(self.press_btn2)
        self.btn3.clicked.connect(self.press_btn3)

        lay = QGridLayout(self)
        lay.addWidget(self.btn1, 0, 0)
        lay.addWidget(self.btn2, 0, 1)
        lay.addWidget(self.btn3, 0, 2)
        lay.addWidget(self.result, 1, 0, 1, 3)

    @pyqtSlot()
    def press_btn1(self):
        command1 = "dir"
        path = "./"
        self.worker.run_command(command1, cwd=path)

    @pyqtSlot()
    def press_btn2(self):
        command2 = "cd"
        path = "./"
        self.worker.run_command(command2, cwd=path, shell=True)

    @pyqtSlot()
    def press_btn3(self):
        command3 = "test.bat"
        path = "./"
        self.worker.run_command(command3, cwd=path, shell=True)

    @pyqtSlot(str)
    def logging(self, string):
        self.result.append(string.strip())


if __name__ == "__main__":
    APP = QApplication(sys.argv)
    ex = TestUI()
    ex.show()
    sys.exit(APP.exec_())

与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…
OGeek|极客中国-欢迎来到极客的世界,一个免费开放的程序员编程交流平台!开放,进步,分享!让技术改变生活,让极客改变未来! Welcome to OGeek Q&A Community for programmer and developer-Open, Learning and Share
Click Here to Ask a Question

1.4m articles

1.4m replys

5 comments

57.0k users

...