如何将持续进程输出管道传输至Python脚本处理并解决挂起问题

我正在开发Python GUI替代团队的命令行传感器数据程序，原程序会把传感器字段和值输出到终端，但输出频繁跳转不够友好。我想把原程序的输出通过管道传给Python脚本，解析后更新GUI。GUI单独运行正常，但管道传输时脚本启动就挂起。已经设置了PYTHONUNBUFFERED环境变量但没用，怀疑是stdin在等待EOF导致的。

示例代码 #

模拟读数的Bash脚本（spammer.sh） #

#!/bin/ksh

while : 
do
  echo ' - Sensor_1_reading: 21'
  sleep 2
  echo ' - Sensor_2_reading: 32'
  sleep 2
  echo ' - Sensor_3_reading: 232'
  sleep 2
done

main-gui.py中的update_function函数 #

def update_function(self):
    print("Running Update!")
    if not sys.stdin.isatty():
        print("Trying to read")
        line = sys.stdin.read()
        print(line)
        
    else:
        print("Nothing on stdin")

执行命令：source spammer.sh | python main-gui.py，程序输出“running update”和“trying to read”后挂起，只有终止模拟脚本时才会正常运行并显示GUI。

1. 逐行读取stdin而非一次性读取 #

sys.stdin.read()会一直等待输入流结束（EOF）才返回，这是脚本挂起的核心原因。改成逐行读取，每收到一行就处理一行：

def update_function(self):
    print("Running Update!")
    if not sys.stdin.isatty():
        print("Trying to read")
        # 逐行读取，实时处理
        for line in sys.stdin:
            line = line.strip()
            if line:
                print(f"Received: {line}")
                # 在这里添加解析逻辑和GUI更新代码
                # 比如提取Sensor_X_reading的数值，更新对应控件
    else:
        print("Nothing on stdin")

注意：如果GUI基于Tkinter这类事件驱动框架，直接在主循环中阻塞读取stdin会导致GUI卡顿，需要把stdin读取放到单独线程中。

2. 用线程处理stdin读取（适配事件驱动GUI） #

以Tkinter为例，创建后台线程监听stdin，避免阻塞主GUI线程：

import sys
import threading
import tkinter as tk

class SensorGUI:
    def __init__(self, root):
        self.root = root
        self.data_label = tk.Label(root, text="传感器数据：")
        self.data_label.pack(pady=20)
        
        # 启动后台线程读取stdin
        self.stdin_thread = threading.Thread(target=self.read_stdin, daemon=True)
        self.stdin_thread.start()
    
    def read_stdin(self):
        for line in sys.stdin:
            line = line.strip()
            if line and "Sensor_" in line and "_reading:" in line:
                # 解析传感器名称和数值
                sensor_info, value = line.split(":")
                sensor_name = sensor_info.strip().replace("- ", "")
                sensor_value = value.strip()
                # 回到主线程更新GUI
                self.root.after(0, self.update_gui, sensor_name, sensor_value)
    
    def update_gui(self, sensor_name, sensor_value):
        self.data_label.config(text=f"{sensor_name}: {sensor_value}")

if __name__ == "__main__":
    root = tk.Tk()
    root.title("传感器监控GUI")
    app = SensorGUI(root)
    root.mainloop()

3. 其他替代方案 #

• 使用命名管道（FIFO）：创建命名管道让传感器程序写入，GUI程序读取，双方可独立启停，比匿名管道更灵活：

  # 创建命名管道
  mkfifo sensor_pipe
  # 后台运行模拟脚本写入管道
  ./spammer.sh > sensor_pipe &
  # 运行GUI程序读取管道
  python main-gui.py < sensor_pipe
  

  Python读取命名管道的方式和stdin一致，同样建议用线程处理。

• 改用subprocess进程通信：如果传感器程序是可执行文件，GUI直接用subprocess.Popen启动它，实时捕获输出，无需外部管道：

  import subprocess
  import threading
  
  def read_sensor_output(proc, update_callback):
      for line in proc.stdout:
          line = line.decode().strip()
          if line:
              update_callback(line)
  
  # 启动传感器程序，捕获标准输出
  proc = subprocess.Popen(["./spammer.sh"], stdout=subprocess.PIPE)
  # 启动线程处理输出，传入GUI更新函数
  threading.Thread(target=read_sensor_output, args=(proc, your_gui_update_func), daemon=True).start()
  

内容的提问来源于stack exchange，提问作者CEEA