파이썬에서 서브 프로세스로 파이썬에서 파이프 쓰기/읽기하기

Question

파이썬에서 파이썬으로 서브 프로세스로 체인을 연결하는 동안 (선행으로 communicate()을 사용하지 않고) 한 줄씩 읽고 쓰는 동안 작동하는 것처럼 보이는 다음 코드가 있습니다. 코드는 유닉스 명령 (mycmd)을 호출하고 출력을 읽은 다음 다른 유닉스 명령의 표준 입력 (next_cmd)에 기록하고 마지막 명령의 출력을 파일로 리디렉션합니다.

# some unix command that uses a pipe: command "a" 
    # writes to stdout and "b" reads it and writes to stdout 
    mycmd = "a | b" 
    mycmd_proc = subprocess.Popen(mycmd, shell=True, 
            stdin=sys.stdin, 
            stdout=subprocess.PIPE, 
            stderr=subprocess.PIPE) 
    # nextCmd reads from stdin, and I'm passing it mycmd's output 
    next_cmd = "nextCmd -stdin" 
    output_file = open(output_filename, "w") 
    next_proc = subprocess.Popen(next_cmd, shell=True, 
            stdin=subprocess.PIPE, 
            stdout=output_file) 
    for line in iter(mycmd.stdout.readline, ''): 
     # do something with line 
     # ... 
     # write it to next command 
     next_proc.stdin.write(line) 
    ### If I wanted to call another command here that passes next_proc output 
    ### line by line to another command, would I need 
    ### to call next_proc.communicate() first? 
    next_proc.communicate() 
    output_file.close() 

이 작동을 표시하고, 그것은 단지 명령의 끝에서 communicate()를 호출합니다.

mycmd1 | mycmd2 | mycmd3 > some_file 

의미 : 나는 당신이 할 수 있도록 다른 명령을 추가하려면이 코드를 확장하기 위해 노력하고있어

, 라인별로 라인을 파이썬에서 mycmd1의 출력을 읽고, 선이,에 공급과정 mycmd2에서 mycmd2의 출력을 읽고 라인 을 처리하여을 처리 한 다음 mycmd3으로 보내고 출력은 some_file이됩니다. 이것이 가능합니까, 아니면 데드락/블로킹/플러쉬되지 않은 버퍼로 끝날 것입니까? 그 사이에 파이썬으로 개입하고 다음 명령으로 넘기기 전에 각 명령의 출력을 한 줄씩 사후 처리하기 때문에 나는 단지 세 개의 유닉스 명령을 파이프로 부르는 것이 아닙니다.

나는 의사 소통을 호출하고 모든 출력을 메모리에로드하는 것을 피하고 싶다. 대신에 한 줄씩 구문 분석하고 싶다. 감사.

import sys 
import subprocess 

def processFirst(out): 
    return out 

def processSecond(out): 
    return out 

def processThird(out): 
    return out 

commands = [("a|b", processFirst), ("nextCmd -stdin", processSecond), ("thirdCmd", processThird)] 

previous_output = None 
for cmd,process_func in commands: 
    if previous_output is None: 
     stdin = sys.stdin 
    else: 
     stdin = subprocess.PIPE 
    proc = subprocess.Popen(cmd, shell=True, 
          stdin = stdin, 
          stdout = subprocess.PIPE) 
    if previous_output is not None: 
     proc.stdin.write(previous_output) 

    out,err = proc.communicate() 
    out = process_func(out) 
    previous_output = out 

은 그냥 출력을 처리해야하는 기능과 함께 명령의 목록에 실행하려는 명령을 추가 :

Answer 1

이 명령의 임의의 수를 처리해야합니다. 마지막 명령의 출력은 루프가 끝날 때 previous_output이됩니다.

교착 상태/버퍼링/기타 문제를 피하려면 proc.communicate()을 사용하여 완료 할 각 명령을 실행하면 결과가 반환됩니다 (예제에서와 같이 직접 읽지 않음). 그런 다음 명령을 완료하기 전에 다음 명령으로 전달합니다.

편집 : 방금 communicate()을 사용하고 싶지 않고 한 줄씩 반응하기를 원합니다. 내 답변을 조금 편집하여

This answer은 select.select()을 사용하여 차단하지 않고 파이프에서 줄 단위로 읽는 방법에 대한 예제를 제공합니다.

import sys 
import subprocess 
import select 
import os 

class LineReader(object): 
    def __init__(self, fd, process_func): 
     self._fd = fd 
     self._buf = '' 
     self._process_func = process_func 
     self.next_proc = None 

    def fileno(self): 
     return self._fd 

    def readlines(self): 
     data = os.read(self._fd, 4096) 
     if not data: 
      # EOF 
      if self.next_proc is not None: 
       self.next_proc.stdin.close() 
      return None 
     self._buf += data 
     if '\n' not in data: 
      return [] 
     tmp = self._buf.split('\n') 
     tmp_lines, self._buf = tmp[:-1], tmp[-1] 
     lines = [] 
     for line in tmp_lines: 
      lines.append(self._process_func(line)) 
      if self.next_proc is not None: 
       self.next_proc.stdin.write("%s\n" % lines[-1]) 

     return lines 

def processFirst(line): 
    return line 

def processSecond(line): 
    return line 

def processThird(line): 
    return line 

commands = [("a|b", processFirst), ("nextCmd -stdin", processSecond), ("thirdCmd", processThird)] 

readers = [] 
previous_reader = None 
for cmd,process_func in commands: 
    if previous_reader is None: 
     stdin = sys.stdin 
    else: 
     stdin = subprocess.PIPE 
    proc = subprocess.Popen(cmd, shell=True, 
          stdin = stdin, 
          stdout = subprocess.PIPE) 

    if previous_reader is not None: 
     previous_reader.next_proc = proc 

    previous_reader = LineReader(proc.stdout.fileno(), process_func) 
    readers.append(previous_reader) 


while readers: 
    ready,_,_ = select.select(readers, [], [], 10.0) 
    for stream in ready: 
     lines = stream.readlines() 
     if lines is None: 
      readers.remove(stream)