Python subprocess PIPE模式下communicate()[0]返回空输出问题
Python subprocess调用i2c_control.exe重定向stdout返回空的修复方案
根因定位
你的代码存在4个直接导致空输出的问题:
- 重复创建子进程:循环逻辑里每次调用
test.application()都会启动一个全新的i2c_control.exe进程,run_read_command内部又单独启动了一个进程,你拿到输出的进程根本没收到预期的read指令,之前启动的进程全部成为僵尸进程挂在后台 communicate()方法逻辑不匹配:该方法会一次性发送输入后,持续等待子进程完全终止才会返回输出。你用的i2c_control.exe是交互式程序,收到read指令执行完后不会自动退出,会持续等待下一条输入,因此communicate()会一直卡在等待进程退出的逻辑,完全不符合交互场景的使用逻辑- 重定向触发的输出缓冲机制:Windows平台下的原生CLI程序,当stdout不直接绑定终端(也就是被PIPE重定向时),会默认开启4KB大小的全缓冲,程序输出的内容会暂存在内存缓冲区,没有凑满缓冲区大小、或者程序主动flush、或者进程退出前,缓冲区内容不会写入管道,自然读不到内容
- 传参逻辑冲突:同时传
shell=True和列表格式的执行命令,在Windows平台会出现命令行解析异常,可能导致子进程根本没有正常启动到可接收指令的状态
排查步骤
- 先放开stderr的打印逻辑,不要注释错误输出,先确认子进程启动过程中有没有抛出找不到命令、权限不足之类的报错
- 手动在终端启动i2c_control.exe,验证两个点:一是程序是否支持直接在命令行后跟指令(比如
i2c_control.exe read 0x10 4)执行完直接退出返回结果,二是如果必须进入交互模式,输入完read指令后输出格式是什么、退出程序需要发送什么指令 - 单独测试启动子进程后不发送任何指令,观察进程是否会自动退出,确认程序的运行模式
修复方案
优先选择非交互模式调用,不需要处理常驻进程、输出缓冲、交互逻辑匹配的问题,踩坑概率最低,修复后的代码如下:
from subprocess import Popen, PIPE # 提前定义好你的register字典 register = { # 替换为实际的寄存器配置 } class TestRegisters(): def run_read_command(self, address, size): # 非交互模式:直接把read指令作为命令行参数传给exe,程序执行完自动退出 cmd = ["i2c_control.exe", "read", str(address), str(size)] # 去掉shell=True,避免命令解析问题,所有管道正常配置 proc = Popen(cmd, bufsize=0, stdout=PIPE, stderr=PIPE, universal_newlines=True) out, err = proc.communicate(timeout=10) # 加超时避免进程卡死 if err: print(f"执行报错: {err}") print(f'output is: {out}\n') return out test = TestRegisters() for key in register: address = key size = int(register[key][1])*4 access = register[key][0] data_type = register[key][2] if access in ('Read', 'Read/Write'): # 删除无意义的重复子进程启动逻辑,直接调用读方法 read = test.run_read_command(address, size)
如果你用的i2c_control.exe不支持命令行直接传参,必须走交互模式,不要用communicate(),改为逐行读写,同时加配置绕过Windows的控制台缓冲:
from subprocess import Popen, PIPE, STARTUPINFO, STARTF_USESHOWWINDOW, SW_HIDE import time class TestRegisters(): def __init__(self): # 初始化时只启动一次子进程,配置启动参数隐藏窗口、绕过缓冲 si = STARTUPINFO() si.dwFlags |= STARTF_USESHOWWINDOW si.wShowWindow = SW_HIDE # creationflags配置CREATE_NO_WINDOW,避免重定向时的缓冲异常 self.proc = Popen( ["i2c_control.exe"], bufsize=0, stdout=PIPE, stdin=PIPE, stderr=PIPE, universal_newlines=True, startupinfo=si, creationflags=0x08000000 ) def run_read_command(self, address, size): cmd = f"read {address} {size}\n" # 交互输入必须加换行符才会被程序识别为提交指令 self.proc.stdin.write(cmd) self.proc.stdin.flush() # 主动刷输入缓冲,确保指令发出去 time.sleep(0.2) # 留够i2c硬件执行时间,根据实际总线速率调整 out = "" # 逐行读输出,直到读到命令提示符或者空内容(终止条件需要根据实际程序的输出规则调整) while True: line = self.proc.stdout.readline() if not line or line.strip().endswith(">"): # 示例假设程序交互提示符是>,按实际情况修改 break out += line print(f'output is: {out}\n') return out def __del__(self): # 用完主动给子进程发退出指令,关闭进程避免后台残留 try: self.proc.stdin.write("exit\n") self.proc.stdin.flush() self.proc.wait(timeout=5) except: self.proc.kill() test = TestRegisters() for key in register: address = key size = int(register[key][1])*4 access = register[key][0] data_type = register[key][2] if access in ('Read', 'Read/Write'): read = test.run_read_command(address, size)
注意:交互模式下的输出读取终止条件需要你根据i2c_control.exe的实际输出规则调整,比如程序执行完read指令后会打印固定的提示符、或者输出固定行数,对应调整读循环的退出逻辑即可。
内容的提问来源于stack exchange,提问作者Manvir
相关产品推荐
相关产品推荐

