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模塊。在任何情況下 - 你知道什麼期望從運行程序，以及它的輸入線是如何結束的。或者你可以在讀取行的時候實現一些超時，所以如果事情不符合預期，可能會引發異常。

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)))