在 Python 中返回一个连续的终端日志

Question

我正在创建一个小应用程序，我可以在其中插入我的 android 手机的 IP 地址，然后按下一个按钮。我通过 ADB 无线连接。这行得通，现在我想在可滚动、不可编辑的文本字段中显示 adb logcat 的日志。但是在这里我遇到了一个问题，我试图像p = os.popen("adb logcat") 这样捕获logcat 命令的输出，然后像print(p.read()) 这样打印它。这只会让我的应用程序（tkinter）冻结，我猜这与打印永远不会结束的事实有关。有没有人知道如何显示 logcat 结果。

函数代码

def logcatcommand():
    p = os.popen("adb logcat")
    print(p.read())

按钮：

button2 = tk.Button(text="Open logcat",width=25, height=3, command=logcatcommand)
button2.grid(row=5, column=0)

你们也知道如何实时显示这些信息吗？我想我必须使用这样的代码：

result = scrolledtext.ScrolledText(window, wrap = tk.WORD, width = 40, height = 10, font = ("Times New Roman", 15))
result.grid(column = 1, row=0)
result.insert(tk.INSERT, output)
result.configure(state ='disabled') 

其中输出是从终端检索的实时数据。

示例行 logcat：

09-10 14:10:28.479   971  3009 I WifiHAL : event received NL80211_CMD_VENDOR, vendor_id = 0x1374, subcmd = 0xd
09-10 14:10:34.779  1526  4567 I BatteryStatsService: In wakeup_callback: resumed from suspend
09-10 14:10:35.321  1526  4567 I BatteryStatsService: In wakeup_callback: suspend aborted

编辑： 感谢 Javier Gonzalez，我能够使用以下方法打印 logcat：

def logcatcommand():
    popen = subprocess.Popen(args="adb logcat", shell=True, stdout=subprocess.PIPE)
    return iter(popen.stdout.readline, b"")


def logcatresult():
    for line in logcatcommand():
        print(line)

但是当尝试将值设置为变量或只是做其他任何事情（例如尝试按下另一个按钮）时，我唯一会看到的是来自 Mac OSX 的彩虹纺车。

问候

Answer 1

我相信这里的提问方式略有不同：Python: How to read stdout of subprocess in a nonblocking way

在您的代码中，我实际上会使用 subprocess 模块而不是 os.popen。

我第一次遇到这个问题时，我使用了上面链接的问题的答案中提到的select 模块。一种更现代的方法是使用 asyncio 代替。问题在于您可以在 SO 中找到使用 asyncio 的多种方法。这些不同是因为 asyncio 在不同版本中发生了显着变化。

编辑#1

您的根本问题是您的主线程正在阻塞。现在你有一个在 logcatresult 函数内永远不会结束的循环，所以它永远不会返回。 Tk 循环永远不会重新获得控制权。这就是它冻结的原因。

一般来说，一个人可以做几件事：

1. 定期进行非阻塞调用。 换句话说：你的主循环需要定期调用一个非阻塞函数来返回任何输出，如果有的话。
2. 将任何阻塞代码移至单独的线程。
3. 使用 asyncio（我不会尝试给你举个例子）

#1：不阻塞地读取日志的想法符合#1 你需要实现周期性调用的逻辑。

#2：如果将阅读循环移动到线程，您可以释放主线程来执行其他任何操作并做出响应。然后让 logcatresult 线程填充您的滚动文本。在这种情况下，你甚至不需要我上面提到的。

以下是可能适合您的简化版本：

from tkinter import *
from tkinter import messagebox
import threading
import subprocess

THREAD = None
PROCESS = None


def read_log(process):
    for line in iter(process.stdout.readline, b""):
        print(line.decode(), end=''),


def read_log_thread():
    """ Button-Event-Handler starting the log reading. """
    global THREAD
    global PROCESS
    PROCESS = subprocess.Popen(["adb logcat"], shell=False, stdout=subprocess.PIPE)
    THREAD = threading.Thread(target=read_log, args=(PROCESS,))
    THREAD.start()


def hello():
    messagebox.showinfo(message='Hello.')


def main():
    root = Tk()
    Button(master=root, text='Start reading', command=read_log_thread).pack()
    buttonX = Button(master=root, text='Hello', command=hello).pack()
    root.mainloop()
    if PROCESS and PROCESS.poll():
        PROCESS.terminate()
    THREAD.join()

if __name__ == '__main__':
    main()

注意事项：

• 此实现只是一个指南。它会有问题（下面会提到一些）。
• 我会尽量不使用全局变量。
• 您需要确保最后正确清理所有内容（所有子进程结束、线程结束等）
• 没有什么可以阻止您按两次按钮，创建多个（泄露的）线程和进程，因为变量是全局的，并且不会进行检查。
• 还请注意，我使用了shell=False，所以最后我可以PROCESS.terminate()。否则，您的程序在进程结束之前不会退出。