python - 如何捕获长时间运行的程序的输出并将其呈现在 Python 的 GUI 中？

Question

我会尽量说得更清楚。

我有一个非常简单的测试脚本来控制电源，该脚本测量来自安捷伦电源+被测单元的一些电流，然后，该脚本打印这些读数，就像这样简单:

PS.write(b"MEAS:CURR? \n")
time.sleep(2)
response = PS.read(1000)
time.sleep(3)
print(response)
(float(response)*1)
E3632A=(float(response)*1)
print (E3632A)

当脚本执行“打印命令”(print (E3632A))时，所有信息都显示到“py.exe”DOS窗口(C:\Windows\py.exe)中。这是我的问题

如何将其嵌入到简单的 GUI 中？我希望我的 GUI 显示 py.exe 正在显示的数据。就这么简单...我已经阅读了互联网上的所有帖子，但没有一个对此有真正的解决方案。

Answer 1

假设您调用的进程是长时间运行的并且不会一次性产生所有输出，这意味着您不能使用 subprocess.Popen.communicate()，因为旨在读取所有输出直至文件末尾。

您将必须使用其他标准技术从管道中读取数据。

由于您希望将其与 GUI 集成，并且该过程需要长时间运行，因此您需要将其输出的读取与 GUI 的主循环协调起来。这让事情变得有些复杂。

TkInter

首先假设您想要使用 TkInter，如您的示例之一所示。这给我们带来了几个问题:

• TkInter 没有与select 模块集成。
• 目前甚至还没有 TkInter 与 asyncio 的规范集成(另请参阅 https://bugs.python.org/issue27546 )。
• 通常建议不要使用 root.update() 组合自定义主循环，让我们通过线程来解决本应基于事件的方法。
• TkInter 的 event_generate() 缺少 Tk 将用户数据与事件一起发送的功能，因此我们无法使用 TkInter 事件将接收到的输出从一个线程传递到另一个线程。

因此，我们将使用线程来解决这个问题(即使我不愿意)，其中主线程控制 Tk GUI，辅助线程读取进程的输出，并且缺少作为 TkInter 中传递数据的 native 方式，我们利用线程安全的队列。

#!/usr/bin/env python3

from subprocess import Popen, PIPE, STDOUT, TimeoutExpired
from threading import Thread, Event
from queue import Queue, Empty
from tkinter import Tk, Text, END


class ProcessOutputReader(Thread):

    def __init__(self, queue, cmd, params=(),
                 group=None, name=None, daemon=True):
        super().__init__(group=group, name=name, daemon=daemon)
        self._stop_request = Event()
        self.queue = queue
        self.process = Popen((cmd,) + tuple(params),
                             stdout=PIPE,
                             stderr=STDOUT,
                             universal_newlines=True)

    def run(self):
        for line in self.process.stdout:
            if self._stop_request.is_set():
                # if stopping was requested, terminate the process and bail out
                self.process.terminate()
                break

            self.queue.put(line)  # enqueue the line for further processing

        try:
            # give process a chance to exit gracefully
            self.process.wait(timeout=3)
        except TimeoutExpired:
            # otherwise try to terminate it forcefully
            self.process.kill()

    def stop(self):
        # request the thread to exit gracefully during its next loop iteration
        self._stop_request.set()

        # empty the queue, so the thread will be woken up
        # if it is blocking on a full queue
        while True:
            try:
                self.queue.get(block=False)
            except Empty:
                break

            self.queue.task_done()  # acknowledge line has been processed


class MyConsole(Text):

    def __init__(self, parent, queue, update_interval=50, process_lines=500):
        super().__init__(parent)
        self.queue = queue
        self.update_interval = update_interval
        self.process_lines = process_lines

        self.after(self.update_interval, self.fetch_lines)

    def fetch_lines(self):
        something_inserted = False

        for _ in range(self.process_lines):
            try:
                line = self.queue.get(block=False)
            except Empty:
                break

            self.insert(END, line)
            self.queue.task_done()  # acknowledge line has been processed

            # ensure scrolling the view is at most done once per interval
            something_inserted = True

        if something_inserted:
            self.see(END)

        self.after(self.update_interval, self.fetch_lines)


# create the root widget
root = Tk()

# create a queue for sending the lines from the process output reader thread
# to the TkInter main thread
line_queue = Queue(maxsize=1000)

# create a process output reader
reader = ProcessOutputReader(line_queue, 'python3', params=['-u', 'test.py'])

# create a console
console = MyConsole(root, line_queue)

reader.start()   # start the process
console.pack()   # make the console visible
root.mainloop()  # run the TkInter main loop

reader.stop()
reader.join(timeout=5)  # give thread a chance to exit gracefully

if reader.is_alive():
    raise RuntimeError("process output reader failed to stop")

由于上述警告，TkInter 代码最终有点偏大。

PyQt

使用PyQt，我们可以大大改善我们的情况，因为该框架已经提供了与子流程集成的 native 方式，以其QProcess类的形式.

这意味着我们可以取消线程并使用 Qt 原生的 Signal 和 Slot 机制。

#!/usr/bin/env python3

import sys

from PyQt5.QtCore import pyqtSignal, pyqtSlot, QProcess, QTextCodec
from PyQt5.QtGui import QTextCursor
from PyQt5.QtWidgets import QApplication, QPlainTextEdit


class ProcessOutputReader(QProcess):
    produce_output = pyqtSignal(str)

    def __init__(self, parent=None):
        super().__init__(parent=parent)

        # merge stderr channel into stdout channel
        self.setProcessChannelMode(QProcess.MergedChannels)

        # prepare decoding process' output to Unicode
        codec = QTextCodec.codecForLocale()
        self._decoder_stdout = codec.makeDecoder()
        # only necessary when stderr channel isn't merged into stdout:
        # self._decoder_stderr = codec.makeDecoder()

        self.readyReadStandardOutput.connect(self._ready_read_standard_output)
        # only necessary when stderr channel isn't merged into stdout:
        # self.readyReadStandardError.connect(self._ready_read_standard_error)

    @pyqtSlot()
    def _ready_read_standard_output(self):
        raw_bytes = self.readAllStandardOutput()
        text = self._decoder_stdout.toUnicode(raw_bytes)
        self.produce_output.emit(text)

    # only necessary when stderr channel isn't merged into stdout:
    # @pyqtSlot()
    # def _ready_read_standard_error(self):
    #     raw_bytes = self.readAllStandardError()
    #     text = self._decoder_stderr.toUnicode(raw_bytes)
    #     self.produce_output.emit(text)


class MyConsole(QPlainTextEdit):

    def __init__(self, parent=None):
        super().__init__(parent=parent)

        self.setReadOnly(True)
        self.setMaximumBlockCount(10000)  # limit console to 10000 lines

        self._cursor_output = self.textCursor()

    @pyqtSlot(str)
    def append_output(self, text):
        self._cursor_output.insertText(text)
        self.scroll_to_last_line()

    def scroll_to_last_line(self):
        cursor = self.textCursor()
        cursor.movePosition(QTextCursor.End)
        cursor.movePosition(QTextCursor.Up if cursor.atBlockStart() else
                            QTextCursor.StartOfLine)
        self.setTextCursor(cursor)


# create the application instance
app = QApplication(sys.argv)

# create a process output reader
reader = ProcessOutputReader()

# create a console and connect the process output reader to it
console = MyConsole()
reader.produce_output.connect(console.append_output)

reader.start('python3', ['-u', 'test.py'])  # start the process
console.show()                              # make the console visible
app.exec_()                                 # run the PyQt main loop

我们最终得到了一些源自 Qt 类的样板，但总体上更简洁。

一般考虑因素

还要确保您调用的进程没有缓冲多个输出行，否则看起来控制台仍然卡住了。

特别是如果被调用者是一个 python 程序，您可以确保它使用 print(...,lush=True) 或使用 python -u callee.py 调用它 强制执行无缓冲输出。