Popenでinteractiveシェルと通信する

Pythonのsubprocess.PopenでインタラクティブなCLIプログラムを起動してテキストベースの通信を行う。

結論とするサンプルから示していく。

サンプル

最小限のインタラクティブなCLIとして次のようなbashプログラムを用意する。

while read -r line; do
    case $line in
    apple) echo red ;;
    banana) echo yellow ;;
    exit) break ;;
    *) echo white ;;
    esac
    echo EOF
done

このプログラムは適当なコマンドapple/banana/exitを受け取ってそれに応じた返答を返す。 Python側でnon-blocking IOを行うことが数行では済まないため返答の終了を表すEOFを毎回最後に出力するようにしている。これをshell.shだとして次のようなPythonプログラムでインタラクティブなbashプログラムと通信ができる。

from subprocess import Popen, PIPE

class Shell():
    def __init__(self, cmd, timeout=5):
        self.process = Popen(cmd, stdin=PIPE, stdout=PIPE, text=True, encoding='utf-8')

    def talk(self, cmd):
        self.process.stdin.write(cmd+'\n')
        self.process.stdin.flush()
        output = ''
        while self.process.poll() == None:
            line = self.process.stdout.readline()
            if line == 'EOF\n':
                break
            output += line
        return output.strip()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.process.poll() == None:
            self.process.send_signal(2)
        return True

with Shell('./shell.sh') as shell:
    print(shell.talk('hello apple'))
    print(shell.talk('hello banana'))
    print(shell.talk('hello tomato'))
    print(shell.talk('exit'))

解説

Popenにstdin=PIPE, stdout=PIPEを指定している。これによってstdin, stdoutに対してwrite/readで書き込むことができる。text=True, encoding='utf-8'を指定することによってバイトストリームが自動的に指定のエンコードでencode/decodeされるため便利である。

参考ページのとおり、writeした際は直後にflushをしてからstdoutを読む方がよいらしい。poll()がNoneを返す時プロセスは起動している。そうでないときプロセスの終了コードを表す。これによってプロセスの生存をチェック可能。send_signal(2)はCtrl-Cの送出を表す。