Sous-processus interactif python communiquez

Question

J'essaie d'apprendre comment écrire une communication interactive de sous-processus.

J'ai besoin de lire stdout et écrire stdin continuellement, voici mon code, il sorte de « marche », mais je ne suis pas sûr si je le fais écrire (il est très Code piraté)

en supposant que je un script appelé app.py comme suit

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

à interactif avec le script ci-dessus (app.py) J'ai quelques code très laid

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

Je ne pense pas que je fais de la bonne façon . quelle est la bonne façon interactive avec un autre script (j'ai besoin stdout, non seulement écrire aveugle à stdin)

Merci

Answer 1

Ce code fonctionnera avec votre app.py, de sorte que vous pouvez obtenir la logique de base de l'interaction de celui-ci . Aussi, je vous suggère de regarder dans le module pexpect. En tout cas - vous DOIT savoir à quoi s'attendre du programme en cours d'exécution et comment cela se termine lignes d'entrée. Ou vous pourriez implémenter un timeout en lisant line, donc il serait possible de lever une exception si quelque chose n'allait pas comme prévu.

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