如何捕获长时间运行的程序的输出并将其呈现在 Python 的 GUI 中? [英] How to capture the output of a long-running program and present it in a GUI in Python?
问题描述
我会尽量说清楚.
我有一个非常简单的测试脚本来控制电源，脚本测量来自安捷伦电源 + 被测单元的一些电流，然后，脚本打印这些读数非常简单:
I have a very simple test script that control a Power Supply, the script measure some current from the Agilent Power Supply + Unit Under Test, then, the script print these readings as simple as:
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)
当脚本执行打印命令"(打印(E3632A))时，所有信息都显示在py.exe"DOS窗口(C:\Windows\py.exe)中.这是我的问题
When the script excecute the "print command" (print (E3632A), all the information is displayed into the "py.exe" DOS Window (C:\Windows\py.exe). Here is my question
如何将其嵌入到简单的 GUI 中?我希望我的 GUI 显示 py.exe 显示的数据.就这么简单......我已经阅读了互联网上的所有帖子，但没有一个真正的解决方案.
How I can embedded this into a simple GUI? I want my GUI display the Data that py.exe is showing. that simple... I have read all post over the internet and none has a real solution to this.
推荐答案
假设您正在调用的进程是长时间运行的并且不会一次性产生所有输出，则意味着您不能使用 subprocess.Popen.communicate()，因为它旨在读取所有输出，直到文件末尾.
Under the assumption that the process you're calling is long-running and doesn't produce all its output in one go, it means you cannot use subprocess.Popen.communicate(), as that is designed to read all output up to an end of file.
您将不得不使用其他标准技术从管道中读取数据.
You will have to use other standard techniques to read from the pipe.
由于您希望将其与 GUI 集成并且该过程需要长时间运行，因此您需要将读取其输出与 GUI 的主循环进行协调.这让事情变得有些复杂.
As you want to integrate it with a GUI and the process is long-running, you will need to coordinate reading its output with the GUI's main loop. This complicates things somewhat.
让我们首先假设您想使用 TkInter，如您的示例之一.这给我们带来了几个问题:
Let's first assume you want to use TkInter, as in one of your examples. That confronts us with a couple of Problems:
- TkInter 没有与 select 模块集成.
- 到目前为止，TkInter 与 asyncio 甚至还没有规范的集成(另见 https://bugs.python.org/issue27546).
- 通常不建议使用
root.update()将自定义主循环组合在一起，让我们通过线程解决本应基于事件的方法. - TkInter 的
event_generate()缺少 Tk 将用户数据与事件一起发送的能力，因此我们不能使用 TkInter 事件将接收到的输出从一个线程传递到另一个线程.
- There's no integration of TkInter with the select module.
- There's even no canonical integration of TkInter with asyncio as of now (also see https://bugs.python.org/issue27546).
- Hacking together a custom main loop using
root.update()is usually recommended against, leaving us solving with threading what should have been an event based approach. - TkInter's
event_generate()is missing Tk's ability to send user data along with the event, so we can't use TkInter events to pass the received output from one thread to another.
因此，我们将使用线程来解决它(即使我不想这样做)，其中主线程控制 Tk GUI，一个辅助线程读取进程的输出，并且缺少TkInter 中传递数据的本机方式，我们使用线程安全的队列.
Thus, we will tackle it with threading (even if I'd prefer not to), where the main thread controls the Tk GUI and a helper thread reads the output from the process, and lacking a native way in TkInter to pass data around, we utilize a thread-safe Queue.
#!/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 代码最终有点偏大.
Due to the aforementioned caveats, the TkInter code ends up a bit on the larger side.
改用 PyQt，我们可以显着改善我们的情况，因为该框架已经以QProcess 类的形式提供了与子流程集成的本机方式.
Using PyQt instead, we can considerably improve our situation, as that framework already comes with a native way to integrate with a subprocess, in the shape of its QProcess class.
这意味着我们可以取消线程，而是使用 Qt 的原生信号和槽机制.
That means we can do away with threads and use Qt's native Signal and Slot mechanism instead.
#!/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 类的样板，但采用了一种整体更简洁的方法.
We end up with a little boilerplate deriving from the Qt classes, but with an overall cleaner approach.
还要确保您正在调用的进程没有缓冲多个输出行，否则它看起来仍然像控制台卡住了一样.
Also make sure that the process you're calling is not buffering multiple output lines, as otherwise it will still look as if the console got stuck.
特别是如果被调用者是一个 python 程序，你可以确保它使用 print(...,flush=True) 或使用 python -u callee.py 调用它 强制无缓冲输出.
In particular if the callee is a python program, you can either ensure that it's using print(..., flush=True) or call it with python -u callee.py to enforce unbuffered output.
这篇关于如何捕获长时间运行的程序的输出并将其呈现在 Python 的 GUI 中?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
