Запустите команду с PyQt5 и получите stdout и stderr - PullRequest
1 голос
/ 11 февраля 2020

Я хочу запустить команду с PyQt5. И я хочу получить stdout и stderr в порядке времени и в реальном времени.

Я разделен на класс пользовательского интерфейса и класс Worker. Существует несколько классов пользовательского интерфейса, но для простоты я указал только один.

Я пытался решить эту проблему, но не могу. Я не могу подключиться между рабочим потоком и функцией регистрации.

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"

Batchfile Проблема

Это результат, когда я запускаю его через командную строку.

Он выводит одну строку каждую секунду в режиме реального времени.

"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"

Ответы [ 2 ]

2 голосов
/ 11 февраля 2020

У вас есть следующие ошибки:

  • Сигналы работают только в объектах QObject, поэтому Worker необходимо наследовать от QObject.

  • Рекомендуется, чтобы QProcess не был членом класса, поскольку мы говорим, что задача 1 выполняется, и, не завершая, вы пытаетесь выполнить задачу 2, так что задача 1 будет заменена, а это не то, что вам нужно, вместо этого можно выполнить QProcess. быть потомком Worker, чтобы ваш жизненный цикл не ограничивался методом, в котором он был создан.

  • Если вы хотите контролировать вывод stderr и stdio отдельно, тогда вам не понравится processChannelMode в QProcess :: MergedChannels, так как это объединит оба выхода, с другой стороны, если вышеперечисленное исключено, вы должны использовать сигнал readyReadStandardError, чтобы узнать, когда stderr изменен.

  • Поскольку QProcess является не является членом класса, трудно получить QProcess в onReadyStandardOutput и onReadyStandardError, bu t для этого вы должны использовать метод 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, поэтому в этом случае вы можете использовать подпроцесс. Откройте, выполнив его в другом потоке и отправив информацию сквозные сигналы:

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_())
0 голосов
/ 11 февраля 2020

Я не совсем уверен, но вы можете попробовать наследовать Worker от QObject или QWidget. Я вполне уверен, что это необходимо для сигналов от пользовательского класса для работы.

...