我想用 PyQt5 运行命令.我想按时间顺序实时获取标准输出和标准错误.
I want to run command with PyQt5. And I want to get the stdout and stderr in time order and in real-time.
我分为 UI 类和 Worker 类.有几个 UI 类，但为简单起见，我只指定了一个.
I separated into UI class and Worker class. There are several UI classes, but for simplicity, I've specified just one.
我试图解决这个问题，但我不能.我无法在 Worker 线程和 logger 函数之间建立连接.
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)
更新:
应用 here 解决方案并进行以下修改仍然会使代码失败:
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"
<小时>
这是我通过命令提示符运行它时的结果.
This is the result when I run it through the command prompt.
每秒实时输出一行.
"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"
这是应用程序的结果.
它在 3 秒后输出整行.而且时间顺序不对.
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"
你有以下错误:
信号只在QObjects中起作用，因此Worker需要继承QObject.
The signals only work in the QObjects so it is necessary for Worker to inherit from QObject.
建议 QProcess 不是该类的成员，因为我们说任务 1 正在执行，而您没有完成就尝试执行任务 2，这样任务 1 将被替换，这不是您想要的，相反，QProcess 可以作为 Worker 的子对象，这样您的生命周期就不会局限于创建它的方法.
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.
如果你想分别监控 stderr 和 stdio 输出，那么你不应该喜欢 processChannelMode 到 QProcess::MergedChannels 因为这将加入两个输出，另一方面，如果上述被消除，那么你必须使用 readyReadStandardError通知 stderr 何时被修改的信号.
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.
由于QProcess不是该类的成员，因此在onReadyStandardOutput和onReadyStandardError中获取QProcess是很困难的，但是为此必须使用sender()方法，该方法具有发出信号的对象.
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.
信号和插槽之间的连接只能建立一次，在您的情况下，您在 press_btn1、press_btn2 和 press_btn3 中执行此操作，因此您将获得 3 倍相同的信息.
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.
不要使用 str，因为它是一个内置函数.
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_())
更新:
QProcess 对执行控制台命令(例如 .bat)有限制，因此在这种情况下，您可以通过在另一个线程中执行 subprocess.Popen 并通过信号发送信息来使用它:
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 运行命令并获取标准输出和标准错误的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持跟版网！