运行 使用 PyQt5 命令并获取标准输出和标准错误
Run command with PyQt5 and getting the stdout and stderr
我想 运行 使用 PyQt5 命令。我想按时间顺序实时获取 stdout 和 stderr。
我分成了UIclass和工人class。有几个 UI class，但为了简单起见，我只指定了一个。
我试过解决这个问题，但我做不到。我无法在工作线程和记录器函数之间建立连接。
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)
更新：
应用并进行以下修改仍然使代码失败：
@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"
批处理文件问题
这是我通过命令提示符运行得到的结果。
实时每秒输出一行
"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秒后输出整行。而且时间顺序不对
"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"
我不是很确定，但是你可以尝试让Worker继承自QObject或QWidget。我相当确定来自用户的信号需要 class 才能工作。
您有以下错误：
信号只在QObject中起作用，所以Worker需要继承QObject。
建议 QProcess 不要成为 class 的成员，因为我们说任务 1 正在执行并且没有完成就尝试执行任务 2，这样任务 1 将replaced 这不是你想要的，相反 QProcess 可以作为 Worker 的子代来完成，这样你的生命周期就不会局限于创建它的方法。
如果你想分别监视 stderr 和 stdio 输出那么你不应该喜欢 processChannelMode 到 QProcess::MergedChannels 因为这将加入两个输出，另一方面如果上面被消除然后您必须使用 readyReadStandardError 信号来了解标准错误何时被修改。
由于QProcess不是class的成员，所以很难在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_())
更新：
QProcess 在执行控制台命令（例如 .bat）方面有限制，因此在这种情况下，您可以通过在另一个线程中执行它并通过信号发送信息来使用 subprocess.Popen：
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。
我分成了UIclass和工人class。有几个 UI class，但为了简单起见，我只指定了一个。
我试过解决这个问题，但我做不到。我无法在工作线程和记录器函数之间建立连接。
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)
更新：
应用
@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"
批处理文件问题
这是我通过命令提示符运行得到的结果。
实时每秒输出一行
"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秒后输出整行。而且时间顺序不对
"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"
我不是很确定，但是你可以尝试让Worker继承自QObject或QWidget。我相当确定来自用户的信号需要 class 才能工作。
您有以下错误：
信号只在QObject中起作用，所以Worker需要继承QObject。
建议 QProcess 不要成为 class 的成员，因为我们说任务 1 正在执行并且没有完成就尝试执行任务 2，这样任务 1 将replaced 这不是你想要的，相反 QProcess 可以作为 Worker 的子代来完成，这样你的生命周期就不会局限于创建它的方法。
如果你想分别监视 stderr 和 stdio 输出那么你不应该喜欢 processChannelMode 到 QProcess::MergedChannels 因为这将加入两个输出，另一方面如果上面被消除然后您必须使用 readyReadStandardError 信号来了解标准错误何时被修改。
由于QProcess不是class的成员，所以很难在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_())
更新：
QProcess 在执行控制台命令（例如 .bat）方面有限制，因此在这种情况下，您可以通过在另一个线程中执行它并通过信号发送信息来使用 subprocess.Popen：
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_())