使用python进行交互式输入/输出

3 回答

• 12

  Linux上针对此问题的两种解决方案：

  第一个是使用文件将输出写入，并同时从中读取：

  from subprocess import Popen, PIPE
  
  fw = open("tmpout", "wb")
  fr = open("tmpout", "r")
  p = Popen("./a.out", stdin = PIPE, stdout = fw, stderr = fw, bufsize = 1)
  p.stdin.write("1\n")
  out = fr.read()
  p.stdin.write("5\n")
  out = fr.read()
  fw.close()
  fr.close()
  

  其次，正如J.F.Sebastian所提出的那样，是使用fnctl模块使p.stdout和p.stder管道无阻塞：

  import os
  import fcntl
  from subprocess import Popen, PIPE  
  def setNonBlocking(fd):
      """
      Set the file description of the given file descriptor to non-blocking.
      """
      flags = fcntl.fcntl(fd, fcntl.F_GETFL)
      flags = flags | os.O_NONBLOCK
      fcntl.fcntl(fd, fcntl.F_SETFL, flags)
  
  p = Popen("./a.out", stdin = PIPE, stdout = PIPE, stderr = PIPE, bufsize = 1)
  setNonBlocking(p.stdout)
  setNonBlocking(p.stderr)
  
  p.stdin.write("1\n")
  while True:
      try:
          out1 = p.stdout.read()
      except IOError:
          continue
      else:
          break
  out1 = p.stdout.read()
  p.stdin.write("5\n")
  while True:
      try:
          out2 = p.stdout.read()
      except IOError:
          continue
      else:
          break
  

  回复于 2024-04-25T18:10:41+08:00
• 3

  目前的答案都不适合我 . 最后，我有这个工作：

  import subprocess
  
  
  def start(executable_file):
      return subprocess.Popen(
          executable_file,
          stdin=subprocess.PIPE,
          stdout=subprocess.PIPE,
          stderr=subprocess.PIPE
      )
  
  
  def read(process):
      return process.stdout.readline().decode("utf-8").strip()
  
  
  def write(process, message):
      process.stdin.write(f"{message.strip()}\n".encode("utf-8"))
      process.stdin.flush()
  
  
  def terminate(process):
      process.stdin.close()
      process.terminate()
      process.wait(timeout=0.2)
  
  
  process = start("./dummy.py")
  write(process, "hello dummy")
  print(read(process))
  terminate(process)
  

  使用此 dummy.py 脚本进行测试：

  #!/usr/bin/env python3.6
  
  import random
  import time
  
  while True:
      message = input()
      time.sleep(random.uniform(0.1, 1.0)) # simulates process time
      print(message[::-1])
  

  需要注意的是：输入/输出始终使用换行符，每次写入后刷新子级的stdin，并使用子级stdout中的 readline() （函数中管理的所有内容） .

  这是一个非常简单的解决方案，在我看来（不是我的，我在这里找到了它：https://eli.thegreenplace.net/2017/interacting-with-a-long-running-child-process-in-python/） . 我使用的是Python 3.6 .

  回复于 2024-04-25T18:10:41+08:00
• 0

  这是一个交互式shell . 你必须在一个单独的线程上运行read（），否则它将阻止write（）

  import sys
  import os
  import subprocess
  from subprocess import Popen, PIPE
  import threading
  
  
  class LocalShell(object):
      def __init__(self):
          pass
  
      def run(self):
          env = os.environ.copy()
          p = Popen('/bin/bash', stdin=PIPE, stdout=PIPE, stderr=subprocess.STDOUT, shell=True, env=env)
          sys.stdout.write("Started Local Terminal...\r\n\r\n")
  
          def writeall(p):
              while True:
                  # print("read data: ")
                  data = p.stdout.read(1).decode("utf-8")
                  if not data:
                      break
                  sys.stdout.write(data)
                  sys.stdout.flush()
  
          writer = threading.Thread(target=writeall, args=(p,))
          writer.start()
  
          try:
              while True:
                  d = sys.stdin.read(1)
                  if not d:
                      break
                  self._write(p, d.encode())
  
          except EOFError:
              pass
  
      def _write(self, process, message):
          process.stdin.write(message)
          process.stdin.flush()
  
  
  shell = LocalShell()
  shell.run()
  

  回复于 2024-04-25T18:10:41+08:00