문제
작업에 대한 좋은 진행률 표시기를 표시하는 명령줄 프로그램(svnadmin verify)에 대한 래퍼 스크립트를 작성하려고 합니다.이를 위해서는 래핑된 프로그램이 출력되는 즉시 각 출력 라인을 볼 수 있어야 합니다.
나는 다음을 사용하여 프로그램을 실행할 것이라고 생각했습니다. 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 (이 문제는 이제 2018 년 8 월 29 일 현재 "폐쇄"되었습니다)
다른 팁
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을 사용하는 경우 입력 메시지가 인쇄되지 않은 경우가 있습니다. 명령으로 시도해보십시오. 인라인 입력이 필요하고 직접 확인하십시오.
하위 프로세스 출력을 스트림으로 직접 지시 할 수 있습니다. 단순화 된 예 :
subprocess.run(['ls'], stderr=sys.stderr, stdout=sys.stdout)
나는 같은 문제를 잠시 후에 달렸다. 내 해결책은 반복을 버리는 것이 었습니다 read
하위 프로세스가 실행되지 않더라도 즉시 반환되는 메소드 등.
실시간 출력 문제 해결 : Python에서 유사한 문제를 겪었고 C 프로그램의 실시간 출력을 캡처했습니다. 나는 "추가"fflush (stdout); "내 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
도움이되기를 바랍니다.
~ Sairam
하위 프로세스의 출력에서 각 바이트에 대해 반복자를 사용할 수 있습니다.이렇게 하면 하위 프로세스에서 인라인 업데이트(' '로 끝나는 줄이 이전 출력 줄을 덮어쓰기)가 가능해집니다.
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
또는 다른 옵션.
(또한보십시오 여기 같은 대답을 위해.)
그만큼 파이썬에서 Asyncio를 사용한 Subprocess Stdin 및 Stdout 스트리밍 블로그 게시물 케빈 맥카시 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())
pexpect 사용 [ 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()