الحصول على الإخراج في الوقت الحقيقي باستخدام العملية الفرعية
-
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 بدلا من القراءة، سيكون هناك بعض الحالات التي لا يتم طباعة رسالة الإدخال. تحاول ذلك مع الأمر يتطلب مدخلا مضمنة وانظر لنفسك.
ويمكنك توجيه الإخراج فرعي أو جانبي للتيارات مباشرة. مثال مبسط:
subprocess.run(['ls'], stderr=sys.stderr, stdout=sys.stdout)
وأنا واجهت نفس المشكلة حظة العودة. وكان بلدي حل للتخلي بالتكرار للأسلوب read
، الذي سيعود على الفور حتى لو لم يتم الانتهاء من فرعي أو جانبي بك المنفذة، وما إلى ذلك.
تم حل مشكلة الإخراج في الوقت الحقيقي:لقد واجهت مشكلة مماثلة في بايثون، أثناء التقاط الإخراج في الوقت الحقيقي من برنامج C.أضفت "فلوش (ستدوت)؛" في كود C الخاص بي.عملت معي.هنا هو رمز القصاصة
<< برنامج سي >>
#include <stdio.h>
void main()
{
int count = 1;
while (1)
{
printf(" Count %d\n", count++);
fflush(stdout);
sleep(1);
}
}
<< برنامج بايثون >>
#!/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.")
واعتمادا على حالة الاستخدام، قد تحتاج أيضا إلى تعطيل التخزين المؤقت في فرعي أو جانبي نفسها.
وإذا كان فرعي أو جانبي سيكون عملية بيثون، يمكنك أن تفعل هذا من قبل الدعوة:
os.environ["PYTHONUNBUFFERED"] = "1"
وأو بدلا من تمرير هذا في حجة env
إلى Popen
.
وإلا، إذا كنت على لينكس / يونكس، يمكنك استخدام أداة stdbuf
. مثلا مثل:
cmd = ["stdbuf", "-oL"] + cmd
وانظر أيضا href="https://unix.stackexchange.com/questions/25372/turn-off-buffering-in-pipe/"> حول stdbuf
أو خيارات أخرى.
و(انظر أيضا href="https://stackoverflow.com/a/52851238/133374"> لنفس الجواب.)
الجري ستدين فرعي أو جانبي والمعياري مع asyncio في بيثون بلوق وظيفة عن طريق كيفن مكارثي يوضح كيفية القيام مع 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 ] مع readlines غير مؤمن سيتم حل هذه المشكلة. انها تنطلق من حقيقة أن الأنابيب يتم تخزينها مؤقتا، وذلك ناتج التطبيق الخاص بك هو الحصول مخزنة بواسطة الأنابيب، وبالتالي لا يمكنك الحصول على هذا الانتاج حتى يملأ المخزن المؤقت أو يموت العملية.
ولقد استخدمت هذا الحل للحصول على الناتج الحقيقي على فرعي أو جانبي. سوف تتوقف هذه الحلقة في أقرب وقت عملية يكمل تاركة حاجة إلى بيان انقطاع أو المحتملة حلقة لا نهائية.
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()
وجدت هذه الوظيفة "التوصيل والتشغيل" <لأ href = "http://www.saltycrane.com/blog/2009/10/how-capture-stdout-in-real-time-python/" يختلط = "نوفولو"> هنا . عملت مثل السحر!
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()
و(وقد تم اختبار هذا الحل مع بيثون 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()