这至少在Python3.4中是有效的

import subprocess

process = subprocess.Popen(cmd_list, stdout=subprocess.PIPE)
for line in process.stdout:
    print(line.decode().strip())

在Python 3.6中，我使用了这个:

import subprocess

cmd = "command"
output = subprocess.call(cmd, shell=True)
print(process)

import time
import sys
import subprocess
import threading
import queue

cmd='esptool.py --chip esp8266 write_flash -z 0x1000 /home/pi/zero2/fw/base/boot_40m.bin'
cmd2='esptool.py --chip esp32 -b 115200 write_flash -z 0x1000 /home/pi/zero2/fw/test.bin'
cmd3='esptool.py --chip esp32 -b 115200 erase_flash'

class ExecutorFlushSTDOUT(object):
    def __init__(self,timeout=15):
        self.process = None
        self.process_output = queue.Queue(0)
        self.capture_output = threading.Thread(target=self.output_reader)
        self.timeout=timeout
        self.result=False
        self.validator=None
        
    def output_reader(self):
        start=time.time()
        while self.process.poll() is None and (time.time() - start) < self.timeout:
            try:
                if not self.process_output.full():
                    line=self.process.stdout.readline()
                    if line:
                        line=line.decode().rstrip("\n")
                        start=time.time()
                        self.process_output.put(line)
                        if self.validator:
                            if self.validator in line: print("Valid");self.result=True

            except:pass
        self.process.kill()
        return
            
    def start_process(self,cmd_list,callback=None,validator=None,timeout=None):
        if timeout: self.timeout=timeout
        self.validator=validator
        self.process = subprocess.Popen(cmd_list,stdout=subprocess.PIPE,stderr=subprocess.PIPE,shell=True)
        self.capture_output.start()
        line=None
        self.result=False
        while self.process.poll() is None:
            try:
                if not self.process_output.empty():
                    line = self.process_output.get()
                if line:
                    if callback:callback(line)
                    #print(line)
                    line=None
            except:pass                
        error = self.process.returncode
        if error:
            print("Error Found",str(error))
            raise RuntimeError(error)
        return self.result

execute = ExecutorFlushSTDOUT()

def liveOUTPUT(line):
    print("liveOUTPUT",line)
    try:
        if "Writing" in line:
            line=''.join([n for n in line.split(' ')[3] if n.isdigit()])
            print("percent={}".format(line))
    except Exception as e:
        pass
    


result=execute.start_process(cmd2,callback=liveOUTPUT,validator="Hash of data verified.")

print("Finish",result)

好吧，我设法解决它没有线程(任何建议，为什么使用线程会更好，感谢)使用从这个问题的一个片段拦截一个子进程的标准输出，而它正在运行

def execute(command):
    process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

    # Poll process for new output until finished
    while True:
        nextline = process.stdout.readline()
        if nextline == '' and process.poll() is not None:
            break
        sys.stdout.write(nextline)
        sys.stdout.flush()

    output = process.communicate()[0]
    exitCode = process.returncode

    if (exitCode == 0):
        return output
    else:
        raise ProcessException(command, exitCode, output)

简单比复杂好。

操作系统库内置模块系统。您应该执行代码并看到输出。

import os
os.system("python --version")
# Output
"""
Python 3.8.6
0
"""

在版本之后，它也打印返回值为0。

这至少在Python3.4中是有效的

import subprocess

process = subprocess.Popen(cmd_list, stdout=subprocess.PIPE)
for line in process.stdout:
    print(line.decode().strip())