Получение вывода в реальном времени с использованием подпроцесса
-
03-07-2019 - |
Вопрос
Я пытаюсь написать сценарий-оболочку для программы командной строки (проверка svnadmin), которая будет отображать хороший индикатор выполнения операции.Для этого мне нужно иметь возможность видеть каждую строку вывода завернутой программы сразу после ее вывода.
Я решил, что просто выполню программу, используя subprocess.Popen
, использовать stdout=PIPE
, затем прочитайте каждую строку по мере ее появления и действуйте соответственно.Однако, когда я запустил следующий код, вывод оказался где-то буферизованным, в результате чего он появился в двух частях: строки с 1 по 332, затем с 333 по 439 (последняя строка вывода).
from subprocess import Popen, PIPE, STDOUT
p = Popen('svnadmin verify /var/svn/repos/config', stdout = PIPE,
stderr = STDOUT, shell = True)
for line in p.stdout:
print line.replace('\n', '')
Немного посмотрев документацию по подпроцессу, я обнаружил bufsize
параметр для Popen
, поэтому я попытался установить bufsize равным 1 (буферизировать каждую строку) и 0 (без буфера), но ни одно из этих значений, похоже, не изменило способ доставки строк.
В этот момент я начал хвататься за соломинку, поэтому написал следующий выходной цикл:
while True:
try:
print p.stdout.next().replace('\n', '')
except StopIteration:
break
но получил тот же результат.
Можно ли получить вывод программы «в реальном времени» из программы, выполняемой с использованием подпроцесса?Есть ли в Python какой-либо другой вариант, совместимый с предыдущими версиями (не exec*
)?
Решение
Я попробовал это, и по какой-то причине код
for line in p.stdout:
...
буферизует агрессивно, вариант
while True:
line = p.stdout.readline()
if not line: break
...
не.Судя по всему это известная ошибка: http://bugs.python.org/issue3907 (Вопрос «Закрыт» по состоянию на 29 августа 2018 г.)
Другие советы
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, bufsize=1)
for line in iter(p.stdout.readline, b''):
print line,
p.stdout.close()
p.wait()
Вы можете попробовать это:
import subprocess
import sys
process = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
while True:
out = process.stdout.read(1)
if out == '' and process.poll() != None:
break
if out != '':
sys.stdout.write(out)
sys.stdout.flush()
Если вы используете readline вместо read, в некоторых случаях входное сообщение не печатается.Попробуйте это с помощью команды, требующей встроенного ввода, и убедитесь сами.
Вы можете направить вывод подпроцесса непосредственно в потоки.Упрощенный пример:
subprocess.run(['ls'], stderr=sys.stderr, stdout=sys.stdout)
Некоторое время назад я столкнулся с той же проблемой.Мое решение состояло в том, чтобы отказаться от итераций для read
метод, который немедленно вернется, даже если ваш подпроцесс еще не завершил выполнение и т. д.
Решена проблема вывода в реальном времени:Я столкнулся с подобной проблемой в Python при захвате вывода программы C в реальном времени.Я добавил "fflush(стандартный вывод);» в моем коде C.Это сработало для меня.Вот фрагмент кода
<< Программа C >>
#include <stdio.h>
void main()
{
int count = 1;
while (1)
{
printf(" Count %d\n", count++);
fflush(stdout);
sleep(1);
}
}
<< Программа Python >>
#!/usr/bin/python
import os, sys
import subprocess
procExe = subprocess.Popen(".//count", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
while procExe.poll() is None:
line = procExe.stdout.readline()
print("Print:" + line)
<< Вывод >> Печать:Считайте 1 Печать:Считайте 2 Печать:Граф 3
Надеюсь, поможет.
~сайрам
Вы можете использовать итератор для каждого байта выходных данных подпроцесса.Это позволяет выполнять встроенное обновление (строки, заканчивающиеся на ' ', перезаписывать предыдущую строку вывода) из подпроцесса:
from subprocess import PIPE, Popen
command = ["my_command", "-my_arg"]
# Open pipe to subprocess
subprocess = Popen(command, stdout=PIPE, stderr=PIPE)
# read each byte of subprocess
while subprocess.poll() is None:
for c in iter(lambda: subprocess.stdout.read(1) if subprocess.poll() is None else {}, b''):
c = c.decode('ascii')
sys.stdout.write(c)
sys.stdout.flush()
if subprocess.returncode != 0:
raise Exception("The subprocess did not terminate correctly.")
В зависимости от варианта использования вы также можете отключить буферизацию в самом подпроцессе.
Если подпроцесс будет процессом Python, вы можете сделать это перед вызовом:
os.environ["PYTHONUNBUFFERED"] = "1"
Или, альтернативно, передайте это в env
аргумент для Popen
.
В противном случае, если вы используете Linux/Unix, вы можете использовать stdbuf
инструмент.Например.нравиться:
cmd = ["stdbuf", "-oL"] + cmd
Смотрите также здесь о stdbuf
или другие варианты.
(Смотрите также здесь за тот же ответ.)
А Потоковая передача stdin и stdout подпроцесса с помощью asyncio в Python сообщение в блоге от Кевин Маккарти показывает, как это сделать с помощью asyncio:
import asyncio
from asyncio.subprocess import PIPE
from asyncio import create_subprocess_exec
async def _read_stream(stream, callback):
while True:
line = await stream.readline()
if line:
callback(line)
else:
break
async def run(command):
process = await create_subprocess_exec(
*command, stdout=PIPE, stderr=PIPE
)
await asyncio.wait(
[
_read_stream(
process.stdout,
lambda x: print(
"STDOUT: {}".format(x.decode("UTF8"))
),
),
_read_stream(
process.stderr,
lambda x: print(
"STDERR: {}".format(x.decode("UTF8"))
),
),
]
)
await process.wait()
async def main():
await run("docker build -t my-docker-image:latest .")
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
Используя ожидание [ http://www.noah.org/wiki/Pexpect ] с неблокирующими строками чтения решит эту проблему.Это связано с тем, что каналы буферизуются, и поэтому выходные данные вашего приложения буферизуются каналом, поэтому вы не можете получить доступ к этому выводу, пока буфер не заполнится или процесс не завершится.
Я использовал это решение для получения вывода в реальном времени в подпроцессе.Этот цикл остановится, как только процесс завершится, исключая необходимость в операторе прерывания или возможном бесконечном цикле.
sub_process = subprocess.Popen(my_command, close_fds=True, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
while sub_process.poll() is None:
out = sub_process.stdout.read(1)
sys.stdout.write(out)
sys.stdout.flush()
Нашел эту функцию «подключи и работай» здесь.Работало как шарм!
import subprocess
def myrun(cmd):
"""from http://blog.kagesenshi.org/2008/02/teeing-python-subprocesspopen-output.html
"""
p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
stdout = []
while True:
line = p.stdout.readline()
stdout.append(line)
print line,
if line == '' and p.poll() != None:
break
return ''.join(stdout)
Полное решение:
import contextlib
import subprocess
# Unix, Windows and old Macintosh end-of-line
newlines = ['\n', '\r\n', '\r']
def unbuffered(proc, stream='stdout'):
stream = getattr(proc, stream)
with contextlib.closing(stream):
while True:
out = []
last = stream.read(1)
# Don't loop forever
if last == '' and proc.poll() is not None:
break
while last not in newlines:
# Don't loop forever
if last == '' and proc.poll() is not None:
break
out.append(last)
last = stream.read(1)
out = ''.join(out)
yield out
def example():
cmd = ['ls', '-l', '/']
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
# Make all end-of-lines '\n'
universal_newlines=True,
)
for line in unbuffered(proc):
print line
example()
Это основной скелет, который я всегда использую для этого.Это упрощает реализацию таймаутов и позволяет справиться с неизбежными зависаниями процессов.
import subprocess
import threading
import Queue
def t_read_stdout(process, queue):
"""Read from stdout"""
for output in iter(process.stdout.readline, b''):
queue.put(output)
return
process = subprocess.Popen(['dir'],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=1,
cwd='C:\\',
shell=True)
queue = Queue.Queue()
t_stdout = threading.Thread(target=t_read_stdout, args=(process, queue))
t_stdout.daemon = True
t_stdout.start()
while process.poll() is None or not queue.empty():
try:
output = queue.get(timeout=.5)
except Queue.Empty:
continue
if not output:
continue
print(output),
t_stdout.join()
(Это решение было протестировано с Python 2.7.15)
Вам просто нужно использовать sys.stdout.flush() после чтения/записи каждой строки:
while proc.poll() is None:
line = proc.stdout.readline()
sys.stdout.write(line)
# or print(line.strip()), you still need to force the flush.
sys.stdout.flush()