>source

나는 이것을 사용하려고 run.py 실행할 스크립트 to_run.py . 이것들은 내가 배우는 데 사용하는 일부 POC 스크립트입니다. 그러나 첫 번째 줄만 제외하고 전체 I/O를 예상했습니다. to_run.py (예 : "HELLOO"), 출력이 없습니다. 이 문제를 어떻게 해결할 수 있습니까?

run.py

#!/usr/bin/python3
from subprocess import PIPE,Popen
from time import sleep
p = Popen(['./to_run.py'],stdin=PIPE,stdout=PIPE,stderr=PIPE)
while True:
    out = p.stdout.readline().decode()
    print(out)
    print("Poll: " + str(p.poll()))

to_run.py

#!/usr/bin/python3
print("HELLOO")
l1 = input("Enter first line: ")
l2 = input("Enter second line: ")
print("First line: "+l1)
print("Second line: "+l2)

  • 답변 # 1

    문제는 당신이 stdout 출력 to_run.py 파이프에 아직 input 기능 to_run.py 프롬프트를 작성해야합니다. stdout 그리고 그것이 문제를 일으키는 것 같습니다. 다음 코드는이를 보여주고 다음 코드에 대한 입력을 제공하여 문제를 해결합니다. input 파이프를 사용하여 stdin . 그만큼 communicate 방법은 입력을 보내는 데 사용됩니다. 나는 또한 지정했다 universal_newlines=True 따라서 출력 데이터는 디코딩 할 필요가없는 문자열입니다.

    import subprocess
    
    p = subprocess.Popen(['./to_run.py'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
    stdout, stderr = p.communicate('Line 1\nLine 2\n')
    print(stdout, end='')
    
    

    최신 정보

    사용하지 않고 할 수 있습니다 communicate 파이프에 직접 쓰고 읽지 만 문서에는 다음과 같은 경고가 있습니다.

    WarningUse communicate() 보다는 .stdin.write , .stdout.read 또는 .stderr.read 다음 중 하나로 인한 교착 상태를 방지하기 위해 다른 OS 파이프 버퍼를 채우고 자식 프로세스를 차단합니다.

    "대화 형"에 관한 한, 제 경험으로는 stdin 데이터를 사전에아니특히 대화 형이라고 생각하는 것입니다. 그러나 최소한 출력이 생성되는대로 처리 할 수 ​​있습니다.

    import subprocess
    
    p = subprocess.Popen(['./to_run.py'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True)
    p.stdin.write('Line 1\n')
    p.stdin.write('Line 2\n')
    p.stdin.close()
    for out in iter(p.stdout.readline, ''): # read rest of output
        print(out, end='')
    p.stdout.close()
    return_code = p.wait()
    
    

    업데이트 2

    원칙적으로 완전히 상호 작용할 수 있습니다. 그러나 당신의 특정 문제는 to_run.py 그 기능인가 input 종료 줄 바꿈없이 프롬프트를 작성하므로 p.stdout.readline()run.py 개행 문자를 기다리고 있습니다. 수정하면 to_run.py 다음과 같이 모든 것이 예상대로 작동합니다.

    to_run.py

    #!/usr/bin/python3
    
    import sys
    print("HELLOO", flush=True)
    print('Enter first line:', flush=True)
    l1 = sys.stdin.readline()
    print('Enter second line:', flush=True)
    l2 = sys.stdin.readline()
    print('First line:', l1, end='')
    print('Second line:', l2, end='')
    
    

    run.py

    import subprocess
    
    cmd = ['./to_run.py']
    p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True)
    print(p.stdout.readline(), end='') # HELLOO
    print(p.stdout.readline(), end='') # Enter first line
    p.stdin.write('Line 1\n')
    p.stdin.flush()
    print(p.stdout.readline(), end='') # Enter second line
    p.stdin.write('Line 2\n')
    p.stdin.flush()
    p.stdin.close()
    for out in iter(p.stdout.readline, ''):
        print(out, end='')
    p.stdout.close()
    return_code = p.wait()
    
    

    다음 코드는 별도의 스레드를 사용하여 stdout 파이프는 사용할 수 없을 때 상황을 처리하는 데 필요한 것 같습니다. readline , 경우와 같이 input 기능이 사용되고 있습니다. 여기서 스레드는 문자를 읽습니다. stdout 한 번에 한 문자 씩 쓰고 Queue 예. 메인 스레드에는 에뮬레이트하기 위해 큐에서 읽는 특수 루틴이 있습니다. readline 그리고 특별한 read_prompt 예상되는 프롬프트를 찾습니다.

    import subprocess
    from threading import Thread
    from queue import Queue
    stdout = Queue()
    def rdr_thread(pipe):
        line = ''
        while True:
            buf = pipe.read(1)
            if not buf:
                stdout.put(None) # show end of file
                return
            stdout.put(buf[0])
            
    def read_line():
        line = ''
        while True:
            ch = stdout.get()
            line += ch
            if ch == '\n':
                return line
    def read_prompt():
        line = ''
        while True:
            ch = stdout.get()
            line += ch
            if line[-2:] == ': ':
                return line
    def output_rest():
        while True:
            ch = stdout.get()
            if ch is None:
                return
            print(ch, end='')
        
    cmd = ['./to_run.py']
    p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True)
    t = Thread(target=rdr_thread, args=(p.stdout,))
    t.start()
    print(read_line(), end='') # HELLOO
    print(read_prompt()) # 'Enter first line: '
    p.stdin.write('Line 1\n')
    p.stdin.flush()
    print(read_prompt()) # 'Enter second line: '
    p.stdin.write('Line 2\n')
    p.stdin.flush()
    p.stdin.close()
    output_rest()
    p.stdout.close()
    return_code = p.wait()
    t.join()
    
    

    시간 초과를 사용한 일반 프롬프트 처리

    import subprocess
    from threading import Thread
    from queue import Queue, Empty
    stdout = Queue()
    eof = False
    def rdr_thread(pipe):
        line = ''
        while True:
            buf = pipe.read(1)
            if not buf:
                stdout.put(None) # show end of file
                eof = True
                return
            stdout.put(buf[0])
            
    def read_prompt():
        """
           read until there seems to be temporarilly no more output
        """
        if eof:
            return ''
        line = ''
        try:
            while True:           
                ch = stdout.get(timeout=.5)
                if ch is None:
                    break
                line += ch
        except Empty:
            pass
        return line
        
    cmd = ['./to_run.py']
    p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True)
    t = Thread(target=rdr_thread, args=(p.stdout,))
    t.start()
    print(read_prompt(), end='')
    p.stdin.write('Line 1\n')
    p.stdin.flush()
    print(read_prompt(), end='')
    p.stdin.write('Line 2\n')
    p.stdin.flush()
    p.stdin.close()
    for chunk in iter(read_prompt, ''):
        print(chunk, end='')
    p.stdout.close()
    return_code = p.wait()
    t.join()
    
    

관련 자료

  • 이전 python - Django Admin 모델은 업데이트 대신 새 인스턴스를 만듭니다
  • 다음 sqlite - 변수를 사용하여 sqlite3 python에 존재하지 않는 경우 테이블 생성