Python：Popen stdout=PIPE无法读取进度条，如何实时获取？

问题 #

我尝试通过UDP Socket传输控制台程序的进度条，实现远程展示任务完成状态，但遇到了问题：subprocess的stdout.PIPE能读取普通print输出，却读不到进度条内容，写入文件或UDP时显示空白。我推测是进度条输出没有以\n结尾，导致没被导入stdout.PIPE——但即使移除普通print语句，进度条仍能在控制台正常显示。

用tqdm库复现问题的代码如下：

sample_process.py #

from time import sleep
from tqdm import tqdm
for i in range(10):
    print(i)
    
for i in tqdm(range(10)):
    sleep(0.5)

client.py #

import socket
from subprocess import Popen, PIPE, CalledProcessError

serverAddressPort   = ("127.0.0.1", 20001)
bufferSize          = 1024

UDPClientSocket = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
cmd = ['python', 'sample_process.py']

with Popen(cmd, stdout=PIPE, bufsize=0, universal_newlines=True) as p:
    for line in p.stdout:
        print(line, end='')
        UDPClientSocket.sendto(line.encode(), serverAddressPort)

if p.returncode != 0:
    raise CalledProcessError(p.returncode, p.args)

server.py #

import socket

UDPServerSocket = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
UDPServerSocket.bind(("127.0.0.1", 20001))
while(True):
    bytesAddressPair = UDPServerSocket.recvfrom(1024)
    message = bytesAddressPair[0]
    print(message.decode())

另外，执行print(p.stdout.read())时，必须等进程结束才会输出内容。请问在Windows 11、Python 3.9.13环境下，有没有办法实时读取进度条？

解决方案 #

针对Windows环境下tqdm进度条无法被subprocess实时读取的问题，可通过以下几种方式解决：

1. 强制tqdm输出到stdout并禁用缓冲 #

tqdm默认会检测输出环境，当检测到非终端（比如PIPE）时可能会调整行为。修改sample_process.py中的tqdm调用，强制指定输出到stdout并关闭缓冲：

from time import sleep
from tqdm import tqdm
import sys

for i in range(10):
    print(i)
    
# 强制输出到stdout，确保非终端环境下正常输出
for i in tqdm(range(10), file=sys.stdout, disable=False, dynamic_ncols=True):
    sleep(0.5)

同时保持client.py中的bufsize=0（无缓冲）配置，确保实时读取字节流。

2. 逐字符读取stdout而非按行读取 #

进度条使用回车\r而非换行\n刷新，按行读取会因缺少换行符而阻塞。修改client.py的读取逻辑，改为逐字节读取：

import socket
from subprocess import Popen, PIPE, CalledProcessError

serverAddressPort   = ("127.0.0.1", 20001)
bufferSize          = 1024

UDPClientSocket = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
cmd = ['python', 'sample_process.py']

with Popen(cmd, stdout=PIPE, bufsize=0, universal_newlines=False) as p:
    while True:
        # 每次读取1字节，捕获进度条的回车刷新信号
        chunk = p.stdout.read(1)
        if not chunk:
            break
        UDPClientSocket.sendto(chunk, serverAddressPort)
        print(chunk.decode(), end='')

if p.returncode != 0:
    raise CalledProcessError(p.returncode, p.args)

这种方式能实时获取进度条的每一次更新，包括\r字符，实现远程同步展示。

3. 修改tqdm输出格式为换行更新 #

如果不需要进度条在控制台原地刷新，可修改tqdm的bar_format，让每次进度更新都输出新行：

for i in tqdm(range(10), bar_format='{l_bar}{bar}| {n_fmt}/{total_fmt}\n'):
    sleep(0.5)

此时进度条每次更新都会生成带\n的输出，client.py的按行读取逻辑就能正常工作，无需修改读取代码。

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