我想用PyQt5运行命令。我想按时间顺序和实时获取stdout和stderr。
我分为UI类和Worker类。有几个UI类,但为简单起见,我仅指定了一个。
我已尝试解决此问题,但不能。我无法在Worker线程和logger函数之间进行连接。
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)
您有以下错误:
信号仅在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
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_())
我不确定,但是您可以尝试让Worker从QObject或QWidget继承。我相当确定这是使用户制作的类发出的信号正常工作所必需的。