利用管道传递命令行输入(stdin)

操作系统进程间通信方式有许多,诸如:

  1. 信号量
  2. 管道
  3. 共享内存

等。

在 Linux 中实现进程间通信相对较为直观容易,而在 Windows 中,并非像 Linux 那样所有系统资源都抽象为文件,为了简单起见,实现进程间通信可借助高级语言所封装的接口。

Python 中有 subprocess 库,可以方便创建子进程,使用包括管道在内的各种技术进行进程间通信。

出于实际需要,考虑做一个 stdin 的 proxy,将命令行或其他地方的 stdin 输入通过管道重定向到子进程。

借鉴 https://pymotw.com/3/subprocess/ 的参考代码:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# For python3

import io
import subprocess

from time import sleep
from sys import platform
from functools import wraps

_python_cmd = platform != 'win32' and 'python3' or 'python'
_subprocess_name = f'{_python_cmd} repeater.py'

def _subprocess(func):
    '''
    _subprocess: subprocess wrap

    Args:
        func: function
    Raise:
        None
    Return:
        None
    '''
    @wraps(func)
    def with_subprocess(is_line_buffering=True):
        print(f'Creating subprocess: "{_subprocess_name}"')

        proc = subprocess.Popen(
            f'{_subprocess_name}',
            shell=True,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
        )
        stdin = io.TextIOWrapper(
            proc.stdin,
            encoding='utf-8',
            line_buffering=is_line_buffering,  # is send data on newline?
        )
        stdout = io.TextIOWrapper(
            proc.stdout,
            encoding='utf-8',
        )

        func(proc, stdin, stdout)

        remainder = proc.communicate()[0].decode('utf-8')
        print(remainder)
        print(f'Subprocess has been completed: "{_subprocess_name}"')

    return with_subprocess



@_subprocess
def show_one_line_at_a_time(proc, stdin, stdout):
    '''
    demo 1: One lone at a time

    Args:
        is_line_buffering: boolean(True)
    '''

    print('One line at a time:')
    for i in range(5):
        line = '{}n'.format(i)
        stdin.write(line)
        output = stdout.readline()
        print(output.rstrip())

        sleep(1)


@_subprocess
def show_all_output_at_once(proc, stdin, stdout):
    '''
    demo 1: All output at once

    Args:
        is_line_buffering: boolean(False)
    '''

    print('All output at once:')
    for i in range(5):
        line = '{}n'.format(i)
        stdin.write(line)

        sleep(1)

    stdin.flush()

@_subprocess
def transfer_data(proc, stdin, stdout):
    '''
    demo 3: Read input from parent process

    Args:
        is_line_buffering: boolean(True)
    '''

    print('Read input from the parent:')
    # for i in range(5):
    while True:
        line = '{}n'.format(input())
        stdin.write(line)
        output = stdout.readline()
        print(output.rstrip())

        sleep(1)


def main():
    show_one_line_at_a_time(True)
    show_all_output_at_once(False)
    transfer_data(True)

if __name__ == '__main__':
    main()

考虑调用此回显程序:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import sys

def repeat():
    '''
    repeat: stdin echoed to stdout

    Args:
        None
    Raise:
        None
    Return:
        None
    '''
    def log_to_file(text, file_name='repeater.log'):
        '''
        log_to_file: log text to file

        Args:
            text: any
            file_name: string('repeater.log')
        Raise:
            None
        Return:
            None
        '''
        with open(file_name, 'a+') as f: 
            f.write(text)
    
    sys.stderr.write('repeater.py: startingn')
    sys.stderr.flush()

    while True:
        next_line = sys.stdin.readline()
        sys.stderr.flush()
        if not next_line:
            break
        sys.stdout.write(next_line)
        sys.stdout.flush()

        log_to_file(next_line)

    sys.stderr.write('repeater.py: exitingn')
    sys.stderr.flush()

def main():
    repeat()

if __name__ == '__main__':
    main()

三个演示分别为:

  1. 分行传送返回输出;
  2. 传送后一次性返回输出;
  3. 命令行代理。

参考:

  1. https://pymotw.com/2/subprocess/ (Python3)
  2. https://pymotw.com/3/subprocess/ (Python2)

作者: YanWen

Web 开发者

发表评论

Fill in your details below or click an icon to log in:

WordPress.com 徽标

You are commenting using your WordPress.com account. Log Out /  更改 )

Google photo

You are commenting using your Google account. Log Out /  更改 )

Twitter picture

You are commenting using your Twitter account. Log Out /  更改 )

Facebook photo

You are commenting using your Facebook account. Log Out /  更改 )

Connecting to %s