うどんコード

スパゲッティからうどんへ

Raging Stream

Pythonでの外部コマンドの実行方法

by ·

Subprocessでのコマンド実行方法いろいろ

コンテンツ 非表示
1 コマンドを実行し、完了まで待機
2 標準出力をキャプチャ
3 標準出力を表示せずに捨てる
4 決められた入力を与える
5 タイムアウトを設定
6 実行完了まで待たない
7 実行途中で終了する
8 インタラクティブに入力を与える

コマンドを実行し、完了まで待機

コマンドを終了まで待機する場合はSubprocess.run()でコマンドを実行します。もっとも単純な使い方はsubprocess.run(args)です。argsには文字列のリストを設定します

from pprint import pprint
import subprocess

def main():
    # 実行完了まで待機 (出力は標準出力に)
    ret = subprocess.run(['ls', '/dev/null'])
    pprint(ret)

実行結果

/dev/null
CompletedProcess(args=['ls', '/dev/null'], returncode=0)

上記のように、終了コードはret.returncodeに格納されます。一般的には0が正常終了、非ゼロがエラーです

標準出力をキャプチャ

標準出力をスクリプトで使用する場合はstdoutオプションを設定します。親プロセスの標準出力には子プロセスの出力は表示されません。encodingの指定を省略した場合はバイト列(b'...')として出力されます

from pprint import pprint
import subprocess

def main():
    # 標準出力をキャプチャ
    ret = subprocess.run(['ping', '127.0.0.1'],
                         stdout=subprocess.PIPE,
                         encoding='utf8')
    pprint(ret)

実行結果

CompletedProcess(args=['ping', '127.0.0.1'], returncode=0, stdout='\nPinging 127.0.0.1 with 32 bytes of data:\nReply from 127.0.0.1: bytes=32 time<1ms TTL=128\nReply from 127.0.0.1: bytes=32 time<1ms TTL=128\nReply from 127.0.0.1: bytes=32 time<1ms TTL=128\nReply from 127.0.0.1: bytes=32 time<1ms TTL=128\n\nPing statistics for 127.0.0.1:\n    Packets: Sent = 4, Received = 4, Lost = 0 (0% loss),\nApproximate round trip times in milli-seconds:\n    Minimum = 0ms, Maximum = 0ms, Average = 0ms\n')

さきほどのret.returncodeに加えて、ret.stdoutにコマンドの標準出力結果が格納されます。

標準出力を表示せずに捨てる

標準出力を単に親プロセスの標準出力に表示したくないだけの場合は、subprocess.PIPEではなく、subprocess.DEVNULLを指定します

from pprint import pprint
import subprocess

def main():
    # 標準出力を表示せずに捨てる
    ret = subprocess.run(['ping', '127.0.0.1'],
                         stdout=subprocess.DEVNULL)
    pprint(ret)

実行結果

CompletedProcess(args=['ping', '127.0.0.1'], returncode=0)

さきほどと違って、ret.stdoutは設定されていません

決められた入力を与える

子プロセスの結果によらず、あらかじめ決められた標準入力を与える場合はinputオプションで指定します

from pprint import pprint
import subprocess

def main():
    # 決められた入力を与える
    ret = subprocess.run(['python', '-i'],
                         input='print("hello")\nexit()',
                         stdout=subprocess.PIPE,
                         encoding='utf8')
    pprint(ret)

実行結果

CompletedProcess(args=['python'], returncode=0, stdout='hello\n')

pythonコマンドを実行し、入力したprint("hello")を実行した結果がret.stdoutに格納されました

タイムアウトを設定

timeoutオプションで子プロセスの終了までの制限時間を設定できます。タイムアウトが発生した場合は、subprocess.TimeoutExpired例外が投げられます。

from pprint import pprint
import subprocess

def main():
    # タイムアウトを設定
    try:
        ret = subprocess.run(['ping', '127.0.0.1'],
                             stdout=subprocess.PIPE,
                             encoding='utf8',
                             timeout=1)  # タイムアウトは1秒
        pprint(ret)
    except subprocess.TimeoutExpired as e:
        pprint(e)
        print(e.output)

実行結果

TimeoutExpired(['ping', '127.0.0.1'], 1)

Pinging 127.0.0.1 with 32 bytes of data:
Reply from 127.0.0.1: bytes=32 time<1ms TTL=128

上記のように、タイムアウトの場合はe.outputにそれまでの標準出力が格納されます

実行完了まで待たない

subprocess.run()ではコマンドの終了を待つことになりますが、subprocess.Popen()の場合、終了を待つのではなく、Popenオブジェクトが返され制御が返されます。その後、親プロセス側でPopen.poll()により終了状況を確認できます (None=実行中、not None=終了)。子プロセス実行結果はrun()の場合と同様に、Popen.returncodeに終了コードが、Popen.stdoutに標準出力結果 (ただし、stringではなくTextIO)が格納されます

from pprint import pprint
import subprocess

def main():
    # 実行完了まで待たない
    p = subprocess.Popen(['ping', '127.0.0.1'],
                         stdout=subprocess.PIPE,
                         encoding='utf8')

    # 完了チェック
    while p.poll() is None:
        print(f"waiting pid:{p.pid} ...")
        time.sleep(1)

    print(f"retcode = {p.returncode}")
    print(p.stdout.read())  # 標準出力取得

実行結果

waiting pid:3320 ...
waiting pid:3320 ...
waiting pid:3320 ...
waiting pid:3320 ...
retcode = 0

Pinging 127.0.0.1 with 32 bytes of data:
Reply from 127.0.0.1: bytes=32 time<1ms TTL=128
Reply from 127.0.0.1: bytes=32 time<1ms TTL=128
Reply from 127.0.0.1: bytes=32 time<1ms TTL=128
Reply from 127.0.0.1: bytes=32 time<1ms TTL=128

Ping statistics for 127.0.0.1:
    Packets: Sent = 4, Received = 4, Lost = 0 (0% loss),
Approximate round trip times in milli-seconds:
    Minimum = 0ms, Maximum = 0ms, Average = 0ms

実行途中で終了する

子プロセスを途中で終了したい場合はPopen.terminate()またはPopen.kill()を呼び出します。 (Windowsの場合は上記は同等、Linuxの場合はterminate()がSIGTERMシグナル、kill()がSIGKILLシグナル)

from pprint import pprint
import subprocess

def main():
    # 実行途中で終了する
    p = subprocess.Popen(['ping', '127.0.0.1'],
                         stdout=subprocess.PIPE,
                         encoding='utf8')
    p.terminate()  # またはp.kill()
    while p.poll() is None:
        print(f"killing pid:{p.pid} ...")
        time.sleep(0.1)

    print(f"retcode = {p.returncode}")
    print(f"stdout = {p.stdout.read()}")

実行結果

killing pid:15008 ...
retcode = 1
stdout =

上記例では、直後にKillしているため標準出力は空になっています。

インタラクティブに入力を与える

subprocess.run(input='*')と違って、subprocess.Popen()の場合は子プロセスの出力結果によって、入力を切り替えるインタラクティブな動作が実現できます。ただし、子プロセスの標準出力を読み込む際にPopen.stdout.readline()とすると、実際に一行出力されるまでブロックされることになるため、これを避けたい場合は別スレッドでstdoutを読み出す少しトリッキーな制御が必要です

from pprint import pprint
import queue
import subprocess
from threading import Thread


def main():
    # インタラクティブに入力を与える
    p = subprocess.Popen(['python', '-i'],
                         stdin=subprocess.PIPE,
                         stdout=subprocess.PIPE,
                         stderr=subprocess.STDOUT,
                         encoding='utf8')

    # 別スレッドでstdoutをqueueに格納するための関数
    def enqueue_output(out, queue):
        for line in iter(out.readline, ''):
            queue.put(line)

    q = queue.Queue()
    t = Thread(target=enqueue_output, args=(p.stdout, q))
    t.start()

    while p.poll() is None:
        try:
            line = q.get_nowait().strip()
            print(f'line: "{line}"')
            if line == '>>>':  # プロンプト出力を待つ
                p.stdin.write('import sys\n')
                p.stdin.write('print(sys.version_info)\n')
                p.stdin.flush()
            elif re.search('sys.version_info', line):
                m = re.search('major=(\d+)', line)
                if int(m.group(1)) < 3:
                    p.stdin.write('exit(-1)\n')
                    p.stdin.flush()
                else:
                    p.stdin.write('exit(0)\n')
                    p.stdin.flush()

        except queue.Empty:
            p.stdin.write('print()\n')
            p.stdin.flush()
            print("no output")
        time.sleep(0.5)

    print(f"retcode = {p.returncode}")

実行結果

no output
line: "Python 3.8.2 (default, Feb 27 2020, 05:27:33)  [GCC 9.2.0 64 bit (AMD64)] on win32"
line: "Type "help", "copyright", "credits" or "license" for more information."
line: ">>>"
line: ">>> >>> sys.version_info(major=3, minor=8, micro=2, releaselevel='final', serial=0)"
retcode = 0

readline()と一行単位で読んでいるため、queueがEmptyの場合に空行をprintする少し煩雑なコードになってしまっています

タグ:

おすすめ