python 대화 형 하위 프로세스 통신

Question

대화식 하위 프로세스 통신을 작성하는 방법을 배우려고합니다.

내가 아래에있는 내 코드는,에서 표준 출력을 읽고 표준 입력 지속적으로 작성해야합니다, 그것은 일종의 "작동"그러나 나는 그것을 내가 가진 가정 (아주 해킹 코드입니다)

작성하고있는 중이 야 확실하지 않다 위의 스크립트 (app.py)와 상호 작용에

import logging 
import random 

def app(): 
    number1 = random.randint(1,100) 
    number2 = random.randint(200,500) 
    logging.info("number1: %s, number2: %s", number1, number2) 
    ans = input("enter sum of {} and {}: ".format(number1, number2)) 
    logging.info("received answer: %s", ans) 
    try: 
     if int(ans) != number1+number2: 
      raise ValueError 
     logging.info("{} is the correct answer".format(ans)) 
    except (ValueError,TypeError): 
     logging.info("{} is incorrect answer".format(ans)) 

def main(): 
    logging.basicConfig(level=logging.DEBUG, filename='log.log') 
    for x in range(10): 
     app() 

if __name__ == '__main__': 
    main() 

을 다음과 같은 app.py라는 스크립트 내가 아주 못생긴 코드

import queue 
import time 
import threading 
import subprocess 
import os 
import pty 
import re 

class ReadStdout(object): 
    def __init__(self): 
     self.queue = queue.Queue() 
     self._buffer_ = [] 

    def timer(self, timeout=0.1): 
     buffer_size = 0 
     while True: 
      if len(self._buffer_) > buffer_size: 
       buffer_size = len(self._buffer_) 
      time.sleep(timeout) 
      if len(self._buffer_) == buffer_size and buffer_size!=0: 
       self.queue.put(''.join(self._buffer_)) 
       self._buffer_ = [] 
       buffer_size = 0 

    def read(self, fd): 
     while True: 
      self._buffer_.append(fd.read(1)) 

    def run(self): 
     timer = threading.Thread(target=self.timer) 
     timer.start() 
     master, slave = pty.openpty() 
     p = subprocess.Popen(['python', 'app.py'], stdout=slave, stderr=slave, stdin=subprocess.PIPE, close_fds=True) 
     stdout = os.fdopen(master) 
     read_thread = threading.Thread(target=self.read, args=(stdout,)) 
     read_thread.start() 
     while True: 
      if self.queue.empty(): 
       time.sleep(0.1) 
       continue 
      msg = self.queue.get() 
      digits = (re.findall('(\d+)', msg)) 
      ans = (int(digits[0])+int(digits[1])) 
      print("got message: {} result: {}".format(msg, ans)) 
      p.stdin.write(b"%d\n" %ans) 
      p.stdin.flush() 

if __name__ == '__main__': 
    x = ReadStdout() 
    x.run() 

내가 느끼지 않는다에게이 나는 올바른 방법으로 일을하고있다 . 다른 스크립트와 상호 작용하는 올바른 방법은 (내가, 표준 출력 표준 입력을뿐만 아니라 시각 장애인 쓰기 필요)

감사

Answer 1

당신이 그것에서 상호 작용의 기본 논리를 얻을 수 있도록이 코드는, 당신의 app.py와 함께 작동합니다 무엇 . 또한 pexpect 모듈을 살펴 보시기 바랍니다. 어떤 경우이든 은 프로그램 실행에서 기대할 수있는 것과 그 입력 행 endswith를 알 수 있어야합니다.. 또는 행을 읽는 동안 시간 초과를 구현할 수 있으므로 예상치 못한 결과가 발생하면 예외를 발생시킬 수 있습니다.

import subprocess 
from functools import partial 

child = subprocess.Popen(['app.py'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True) 

# iterate while child is not terminated 
while child.poll() is None: 
    line = '' 
    # read stdout character by character until a colon appears 
    for c in iter(partial(child.stdout.read, 1), ''): 
     if c == ':': 
      break 
     line += c 
    if "enter sum" in line: 
     numbers = filter(str.isdigit, line.split()) 
     numbers = list(map(int, numbers)) 
     child.stdin.write("{0}\n".format(sum(numbers))) 

([(원시 \의 _input에 의해) 사용자 상호 작용을 필요로 다른 파이썬 스크립트를 실행 한 스크립트에서 파이썬의 하위 프로세스와는 popen을 사용]의