要从我的python脚本启动程序，我使用以下方法:
def execute(command):
process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
output = process.communicate()[0]
exitCode = process.returncode
if (exitCode == 0):
return output
else:
raise ProcessException(command, exitCode, output)
所以当我启动一个进程，比如进程。执行("mvn clean install")时，我的程序会一直等待，直到进程完成，只有到那时我才能得到程序的完整输出。这是恼人的，如果我正在运行一个进程，需要一段时间才能完成。
我能让我的程序一行一行地写进程输出吗，在循环结束之前轮询进程输出什么的?
我找到了这篇文章，可能与此有关。
当前回答
在Python中>= 3.5使用subprocess.run为我工作:
import subprocess
cmd = 'echo foo; sleep 1; echo foo; sleep 2; echo foo'
subprocess.run(cmd, shell=True)
(在没有shell=True的情况下，在执行过程中获得输出也可以工作) https://docs.python.org/3/library/subprocess.html#subprocess.run
其他回答
在Python中>= 3.5使用subprocess.run为我工作:
import subprocess
cmd = 'echo foo; sleep 1; echo foo; sleep 2; echo foo'
subprocess.run(cmd, shell=True)
(在没有shell=True的情况下，在执行过程中获得输出也可以工作) https://docs.python.org/3/library/subprocess.html#subprocess.run
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)
可以在命令输出后立即使用iter对行进行处理。readline”、“)。下面是一个完整的例子，展示了一个典型的用例(感谢@jfs的帮助):
from __future__ import print_function # Only Python 2.x
import subprocess
def execute(cmd):
popen = subprocess.Popen(cmd, stdout=subprocess.PIPE, universal_newlines=True)
for stdout_line in iter(popen.stdout.readline, ""):
yield stdout_line
popen.stdout.close()
return_code = popen.wait()
if return_code:
raise subprocess.CalledProcessError(return_code, cmd)
# Example
for path in execute(["locate", "a"]):
print(path, end="")
实际上有一个非常简单的方法来做到这一点，当你只想打印输出:
import subprocess
import sys
def execute(command):
subprocess.check_call(command, shell=True, stdout=sys.stdout, stderr=subprocess.STDOUT)
在这里，我们只是将子进程指向我们自己的标准输出，并使用现有的成功或异常api。
简单比复杂好。
操作系统库内置模块系统。您应该执行代码并看到输出。
import os
os.system("python --version")
# Output
"""
Python 3.8.6
0
"""
在版本之后，它也打印返回值为0。
