اجرای یک پروسه و گرفتن خروجی در پایتون

در این مطلب می‌گیم که چطور میشه در پایتون یک پروسه دیگر را اجرا کرد و خروجی استاندارد و ورودی استاندارد رو بگیریم و ازش استفاده بکنیم. خروجی و ورودی استاندارد همون چیزایی هستن که تو محیط متنی چاپ میشن یا کاربر توی ورودی برنامه وارد می‌کنه. در واقع توی این مطلب یاد می‌گیرید که چطور می‌تونید در پایتون با برنامه های کنسولی دیگه تعامل کنید.

پایپ (pipe) چیست؟

به طور پیش‌فرض سیستم‌عامل ورودی‌ها رو از موس و کیبورد می‌گیره و خروجی‌ها رو روی صفحه‌نمایش می‌نویسه. اما در بعضی مواقع نیاز هست که یک برنامه از خروجی‌های یک برنامه (یا دستور) دیگه استفاده کنه یا به ورودی استاندارد یک برنامه داده ارسال کنه. در چنین شرایطی pipe استفاده میشه. pipe یک فضای موقتی در حافظه برای جابه‌جایی اطلاعات بین دو برنامه هست که البته یک طرفه هم هست؛ یعنی مثلا برای گرفتن خروجی باید از یک pipe و برای نوشتن ورودی هم از یک pipe دیگر باید استفاده کرد.

اجرای یک پروسه در پایتون

ماژولی که ما برای این کار استفاده می‌کنیم ماژول subprocess هست که یک ماژول توکار پایتونی هست. برای این کار دو روش وجود داره، یکی استفاده از تابع run و یکی هم استفاده از کلاس Popen.

تابع run بیشتر به درد اجرای یک دستور می‌خوره، یه دستور یا برنامه رو اجرا می‌کنه اگر خواستید ورودی ها رو ارسال می‌کنه و خروجی‌ها رو تا بسته شدن برنامه یا دستور ذخیره می‌کنه و در نهایت یک نمونه (instance) از CompletedProcess بر می‌گرداند. این تابع توی برنامه‌تون وقفه ایجاد می‌کنه و تا پایان کار برنامه منتظر می‌مونه.

در حالی که کلاس Popen یک پروسه ایجاد می‌کند و به طور همزمان به روند ادامه می‌دهد. در واقع تابع run خودش ازکلاس Popen استفاده می‌کنه، خروجی ها رو ذخیره و ورودی ها رو با استفاده از Popen.communicate() ارسال می‌کنه (این تابع تا پایان پروسه وقفه ایجاد می‌کنه). برای این که با کلاس Popen کار کنید و خروجی بگیرید و ورودی‌ها رو بگیرید بهترین کار این هست که به شکل چند ریسمانی برنامه بنویسید. در این مطلب با کلاس Popen کار می‌کنیم.

نکته‌ای که خوبه بدونید این هست که کلا پایتون single-thread هست، این یعنی حتی اگر اسکریپت شما به صورت multi-thread باشه پایتون فقط از یک پروسه و از یک ترد (و یک هسته (مجازی) cpu) استفاده می‌کنه و ترد ها از وقفه های همدیگه استفاده می‌کنن. برای این که بتونید به معنای واقعی از چند هسته توی پایتون استفاده کنید یکی از روش ها multi-processing هست که در پایتون ماژول multiprocessing برای این کار بهتره.

کلاس Popen

پارامتر های تابع run و کلاس Popen مثل هم هستن چون همونطور که گفتم تابع run خودش داره از همین کلاس استفاده می‌کنه.

class subprocess.Popen(args, bufsize=- 1, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=None, startupinfo=None, creationflags=0, restore_signals=True, start_new_session=False, pass_fds=(), *, group=None, extra_groups=None, user=None, umask=- 1, encoding=None, errors=None, text=None, pipesize=- 1, process_group=None)

کلاس Popen (در زمان تعریف یک instance) یک پروسه فرزند را اجرا می‌کند و وقفه‌ای هم در روند برنامه ایجاد نمی‌کنه.

پارامتر args می‌تواند یک رشته و یا یک دنباله (از دستور یا برنامه و پارامتر ها) و یا یک path-like object باشه.

from subprocess import Popen

Popen(["git", "commit", "-m", "bugfix"])

پارامتر های stdin, stdout و stderr در واقع مشخص کننده سه استریم استاندارد پروسه هست که به ترتیب خروجی استاندارد، ورودی استاندارد و خروجی خطای استاندارد هستند.

برای استفاده از pipe متغیر (ثابت) PIPE را از ماژول subprocess وارد کنید.

from subprocess import Popen, PIPE

proc = Popen("cmd", stdin = PIPE, stdout = PIPE, stderr = PIPE, encoding='utf-8')

اگر خواستید stderr را هم در stdout دریافت کنید می‌تونید ثابت STDOUT را نیز وارد کنید و آن را برای استریم خطا بنویسید: Popen(..., stderr = STDOUT)

در پایتون می‌تونید با کد زیر در خروجی استاندارد خطا (stderr) هر متنی می‌خواهید چاپ کنید. البته exception های پایتون به طور پیشفرض در stderr نوشته می‌شوند.

from sys import stderr
print("Hello nether dimension", file = stderr)

پارامتر encoding تعیین کننده انکودر خروجی‌ها و ورودی‌ها هست، اگر این پارامتر تعیین نشه خروجی و ورودی به صورت باینری هست. اگر قصد دارید به صورت متنی با برنامه کار کنید می‌تونید انکودینگ رو UTF-8 یا ANSI و … تعیین کنید.

به شکل ساده اگر بخوایید توی ورودی بنویسید و خروجی رو بخونید می‌تونید به شکل زیر عمل کنید:

from time import sleep
from subprocess import Popen, PIPE, DEVNULL

proc = Popen(["git", "commit", "-F", "-"],
             stdout = PIPE,
             stdin = PIPE,
             stderr = DEVNULL,
             encoding="utf-8")
proc.stdin.write("foo bar\n")
proc.stdin.close()        # Write EOF to stdin stream
proc.wait()                  # wait for git to do his job
print("output is:", proc.stdout.read(), "="*20, sep='\n')

برنامه‌هایی که وقفه دارند

مثال بالا یک دستور بود که به سادگی اجرا شد و پس از انجام کار خاتمه پیدا کرد. ولی در بسیاری مواقع زمانی که یک برنامه را اجرا می‌کنید منتظر ورودی می‌ماند (مثل تابع input در پایتون) در چنین مواقعی اگر سعی کنید خروجی استاندارد را بخوانید برنامه شما بلوکه می‌شود، یعنی توسط پروسه اجرا شده در استریم یک وقفه ایجاد می‌شود که تا زمانی که ورودی استاندارد وارد شود ادامه دارد. در حالتی که ورودی استاندراد pipe شده باشد برنامه شما است که باید به آن ورودی دهد که آن هم بلاکه شده و تمام، مجبورید برنامه را terminate کنید.

برای حل این مشکل می توانید مسئولیت خاندن را به یک ریسمان (Thread) دیگر محول کنید. یعنی یک ریسمان تولیدکننده بسازید و در ریسمان اصلی مصرف کنید!

عموما برای پیاده‌سازی برنامه‌نویسی چندریسمانی تولیدکننده و مصرف کننده از صف (Queue) استفاده میشه برای همین اگر از پایتون ۲ استفاده می‌کنید از ماژول Queue و در پایتون ۳ از ماژول queue کلاس Queue را وارد کنید.

try:
    from queue import Queue, Empty
except ImportError:
    from Queue import Queue, Empty

کد بالا مشخصه چیکار می‌کنه! شما می‌تونید ساده‌ش کنید و یکی از دو حالت رو بنویسید. برنامه‌ای که من داشتم می‌نوشتم معلوم نبود دقیقا رو کدوم نسخه قراره اجرا بشه!

def enqueue_output(out, q):
    for line in iter(out.readline, ""):
        q.put_nowait(line)
    out.close()

حالا بعد از اجرای پروسه رسمان خواننده را اجرا کنید. می‌توانید مانند کد زیر یک تابع تعریف کنید.

from subprocess import Popen, PIPE
from queue import Queue, Empty
from threading import Thread

def enqueue_output(out, q):
    for line in iter(out.readline, ""):
        q.put_nowait(line)
    out.close()


def exec_proc(args:list[str] | str):
    stdout_q, stderr_q = Queue(), Queue()
    proc = Popen(args, stdout=PIPE, stderr=PIPE, stdin=PIPE)
    out_th = Thread(target=enqueue_output, args=(proc.stdout, stdout_q), daemon=True)
    err_th = Thread(target=enqueue_output, args=(proc.stderr, stderr_q), daemon=True)
    out_th.start()
    err_th.start()
    return proc, stdout_q, stderr_q


def readline(q, timeout=1) -> str | None:
    try:
        line = q.get(True, timeout)
        return line
    except Empty:
        return None

حالا هر بار که می‌خواهید یک پروسه ایجاد کنید می‌توانید تابع exec_proc را فراخوانی کنید، خروجی های تابع به ترتیب یک نمونه (instance) از Popen و تو نمونه از queue خواهد بود که یکی برای خروجی استاندارد و یکی هم برای خطای استاندارد هست. برای خواندن خروجی می‌تونید از تابع readline استفاده کنید. پارامتر اول صف مورد نظر و پارامتر اختیاری دوم هم مدت زمان timeout هست؛ این پارامتر تعیین می‌کنه که چه مدتی باید منتظر یمونه تا خروجی رو بگیره، توی except می‌تونین هر چی که می‌خوایین بنویسین ولی اینجا None بر می‌گردونه.

برای مثال می‌تونید در اینجا پروژه‌ای که به دانشجو های مبانی برای پایان‌ترم نیم‌سال اول ۱۴۰۱ دادم رو ببینید! توی این پروژه برای اولین بار خودم آستین بالا زدم و یک برنامه ساده برای تصحیح پروژه هاشون نوشتم. فایل checker.py اسکریپت دانشجو ها رو اجرا می‌کنه، ورودی میده و خورجی رو چک می‌کنه. فایل tester.py هم نسخه ساده‌تر checker هست که به دانشجو ها داده شده بود تا برنامه‌شون رو دیباگ کنن.