使用 PyQt5 运行命令并获取 stdout 和 stderr [英] Run command with PyQt5 and getting the stdout and stderr
问题描述
我想用 PyQt5 运行命令.我想按时间顺序实时获取标准输出和标准错误.
我分为 UI 类和 Worker 类.有多个 UI 类，但为简单起见，我只指定了一个.
我试图解决这个问题，但我不能.我无法在 Worker 线程和 logger 函数之间进行连接.
test_ui.py
导入系统导入子流程从 PyQt5.QtWidgets 导入 QApplication、QWidget、QHBoxLayout、QVBoxLayout从 PyQt5.QtWidgets 导入 QPushButton、QTextEdit从工人进口工人类 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()定义 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)自我展示()def press_btn1(self):命令1 =目录"路径 = "./"self.worker.run_command(command1, path)self.worker.outSignal.connect(self.logging)def press_btn2(self):命令2 =CD"路径 = "./"self.worker.run_command(command2, path)self.worker.outSignal.connect(self.logging)def press_btn3(self):command3 = "whoami"路径 = "./"self.worker.run_command(command3, path)self.worker.outSignal.connect(self.logging)定义日志记录(自我，str):self.result.append(str.strip())如果 __name__ == "__main__":APP = QApplication(sys.argv)ex = TestUI()sys.exit(APP.exec_())worker.py
从 PyQt5.QtCore 导入 QProcess，pyqtSignal班级工人:outSignal = pyqtSignal(str)错误信号 = pyqtSignal(str)def __init__(self):self.proc = QProcess()def run_command(self, cmd, path):self.proc.setWorkingDirectory(路径)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):结果 = self.proc.readAllStandardOutput().data().decode()self.outSignal.emit(结果)def onReadyStandardError(self):结果 = self.proc.readAllStandardError().data().decode()self.errSignal.emit(结果)更新:
应用此处的解决方案并进行以下修改后代码仍然失败:
@pyqtSlot()def press_btn1(self):命令1 =目录"路径 = "./"self.worker.run_command(command1, path)@pyqtSlot()def press_btn2(self):命令2 =CD"路径 = "./"self.worker.run_command(command2, path)@pyqtSlot()def press_btn3(self):command3 = "test.bat"路径 = "./"self.worker.run_command(command3, path)@pyqtSlot(str)定义日志(自我，味精):msg = msg.strip()如果 msg != "":self.result.append(msg)test.bat
@echo off回声输出1"超时/t 11>&2 回显错误 1"超时/t 1回声输出2"超时/t 11>&2 回显错误 2"<小时>
批处理文件问题
这是我通过命令提示符运行时的结果.
它每秒实时输出一行.
输出 1"等待0秒，按一个键继续...错误 1"等待0秒，按一个键继续...输出2"等待0秒，按一个键继续...错误2"这是申请的结果.
它在 3 秒后输出整行.而且时间顺序不对.
输出 1"等待1秒，按一个键继续...0等待1秒，按一个键继续...0输出2"等待1秒，按一个键继续...0错误 1"错误2"您有以下错误:
信号只在 QObject 中起作用，因此 Worker 必须从 QObject 继承.
建议 QProcess 不是类的成员，因为我们说任务 1 正在执行并且没有完成你尝试执行任务 2，这样任务 1 将被替换，这不是你想要的，相反，QProcess 可以作为 Worker 的子进程完成，这样您的生命周期就不会局限于创建它的方法.
如果你想分别监控 stderr 和 stdio 输出，那么你不应该喜欢 processChannelMode 到 QProcess::MergedChannels 因为这将连接两个输出，另一方面，如果上述被消除，那么你必须使用 readyReadStandardError用于知道何时修改 stderr 的信号.
由于 QProcess 不是该类的成员，因此很难在 onReadyStandardOutput 和 onReadyStandardError 中获取 QProcess，但为此您必须使用具有发出信号的对象的 sender() 方法.
信号和槽之间的连接应该只建立一次，在你的情况下，你在 press_btn1、press_btn2 和 press_btn3 中这样做，所以你会得到 3 倍相同的信息.
不要使用
str，因为它是一个内置函数.
综合以上，解决办法是:
worker.py
from PyQt5.QtCore import QObject, QProcess, pyqtSignal, pyqtSlot类工人(QObject):outSignal = pyqtSignal(str)错误信号 = pyqtSignal(str)def run_command(self, cmd, path):proc = QProcess(self)proc.setWorkingDirectory(路径)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()结果 = proc.readAllStandardOutput().data().decode()self.outSignal.emit(结果)@pyqtSlot()def onReadyStandardError(self):proc = self.sender()结果 = proc.readAllStandardError().data().decode()self.errSignal.emit(结果)test_ui.py
导入系统从 PyQt5.QtCore 导入 pyqtSlot从 PyQt5.QtWidgets 导入 QApplication、QGridLayout、QPushButton、QTextEdit、QWidget从工人进口工人类 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()定义 init_ui(self):self.btn1.clicked.connect(self.press_btn1)self.btn2.clicked.connect(self.press_btn2)self.btn3.clicked.connect(self.press_btn3)躺 = QGridLayout(自我)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):命令1 =目录"路径 = "./"self.worker.run_command(command1, path)@pyqtSlot()def press_btn2(self):命令2 =CD"路径 = "./"self.worker.run_command(command2, path)@pyqtSlot()def press_btn3(self):command3 = "whoami"路径 = "./"self.worker.run_command(command3, path)@pyqtSlot(str)定义日志(自我，字符串):self.result.append(string.strip())如果 __name__ == "__main__":APP = QApplication(sys.argv)ex = TestUI()例如.show()sys.exit(APP.exec_())更新:
QProcess 对执行控制台命令(例如 .bat)有限制，因此在这种情况下，您可以通过在另一个线程中执行 subprocess.Popen 并通过信号发送信息来使用它:
worker.py
导入子流程进口螺纹从 PyQt5 导入 QtCore类工人(QtCore.QObject):outSignal = QtCore.pyqtSignal(str)def run_command(self, cmd, **kwargs):线程.线程(target=self._execute_command, args=(cmd,), kwargs=kwargs, daemon=True).开始()def _execute_command(self, cmd, **kwargs):proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kwargs)对于 proc.stdout 中的行:self.outSignal.emit(line.decode())test_ui.py
导入系统从 PyQt5.QtCore 导入 pyqtSlot从 PyQt5.QtWidgets 导入 QApplication、QGridLayout、QPushButton、QTextEdit、QWidget从工人进口工人类 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()定义 init_ui(self):self.btn1.clicked.connect(self.press_btn1)self.btn2.clicked.connect(self.press_btn2)self.btn3.clicked.connect(self.press_btn3)躺 = QGridLayout(自我)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):命令1 =目录"路径 = "./"self.worker.run_command(command1, cwd=path)@pyqtSlot()def press_btn2(self):命令2 =CD"路径 = "./"self.worker.run_command(command2, cwd=path, shell=True)@pyqtSlot()def press_btn3(self):command3 = "test.bat"路径 = "./"self.worker.run_command(command3, cwd=path, shell=True)@pyqtSlot(str)定义日志(自我，字符串):self.result.append(string.strip())如果 __name__ == "__main__":APP = QApplication(sys.argv)ex = TestUI()例如.show()sys.exit(APP.exec_())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"
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
strsince 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_())
这篇关于使用 PyQt5 运行命令并获取 stdout 和 stderr的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
