Python中的基本终端模拟

Question

我一直在试图编写一个基本的终端仿真脚本，因为某些原因，我在我的Mac上没有终端访问权限。但是要在搅拌机中编写游戏引擎脚本，控制台通常在您开始搅拌的终端中打开，这一点至关重要。
只是做简单的事情，如删除，重命名等我曾经执行命令使用stream = os.popen(command)，然后print (stream.read())。这对大多数事情都很好，但不适用于任何交互式事物。
我很快就发现了一种新的方式：
sp = subprocess.Popen(["/bin/bash", "-i"], stdout = subprocess.PIPE, stdin = subprocess.PIPE, stderr = subprocess.PIPE)然后print(sp.communicate(command.encode()))。那应该产生一个我可以像终端一样使用的交互式shell，不是吗？

但无论哪种方式，我不能保持连接，并使用最后一个例子，我可以打电话sp.communicate一次，给我下面的输出（在这种情况下为“LS /”）和一些错误：
(b'Applications\n[...]usr\nvar\n', b'bash: no job control in this shell\nbash-3.2$ ls /\nbash-3.2$ exit\n')。 第二次它给我一个ValueError: I/O operation on closed file. 有时（像'ls'）我只会得到这个错误：b'ls\nbash-3.2$ exit\n'。

这是什么意思？我怎样才能模拟一个终端与python，使我可以控制一个交互式shell或运行搅拌机和控制台通信？

Answer 1

假设你想要的不断要求输入一个交互的shell，你可以尝试以下方法：

import subprocess 
import re 

while True: 
    # prevents lots of python error output 
    try: 
     s = raw_input('> ') 
    except: 
     break 

    # check if you should exit 
    if s.strip().lower() == 'exit': 
     break 

    # try to run command 
    try: 
     cmd = subprocess.Popen(re.split(r'\s+', s), stdout=subprocess.PIPE) 
     cmd_out = cmd.stdout.read() 

     # Process output 
     print cmd_out 

    except OSError: 
     print 'Invalid command' 

Answer 2

看来，你应该对新的控制终端运行它，用“forkpty”分配。 要抑制警告“无作业控制...”，您需要调用“setsid”。

Answer 3

这里是我努力做的，你想在Windows中做什么..一个更难的问题，因为Windows不遵循任何标准，但他们自己。对这段代码进行微小的修改应该可以让你确切地知道你在找什么。

''' 
Created on Mar 2, 2013 

@author: rweber 
''' 
import subprocess 
import Queue 
from Queue import Empty 
import threading 


class Process_Communicator(): 

    def join(self): 
     self.te.join() 
     self.to.join() 
     self.running = False 
     self.aggregator.join() 

    def enqueue_in(self): 
     while self.running and self.p.stdin is not None: 
      while not self.stdin_queue.empty(): 
       s = self.stdin_queue.get() 
       self.p.stdin.write(str(s) + '\n\r') 
      pass 

    def enqueue_output(self): 
     if not self.p.stdout or self.p.stdout.closed: 
      return 
     out = self.p.stdout 
     for line in iter(out.readline, b''): 
      self.qo.put(line) 

    def enqueue_err(self): 
     if not self.p.stderr or self.p.stderr.closed: 
      return 
     err = self.p.stderr 
     for line in iter(err.readline, b''): 
      self.qe.put(line) 

    def aggregate(self): 
     while (self.running): 
      self.update() 
     self.update() 

    def update(self): 
     line = "" 
     try: 
      while self.qe.not_empty: 
       line = self.qe.get_nowait() # or q.get(timeout=.1) 
       self.unbblocked_err += line 
     except Empty: 
      pass 

     line = "" 
     try: 
      while self.qo.not_empty: 
       line = self.qo.get_nowait() # or q.get(timeout=.1) 
       self.unbblocked_out += line 
     except Empty: 
      pass 

     while not self.stdin_queue.empty(): 
       s = self.stdin_queue.get() 
       self.p.stdin.write(str(s) + '\n\r') 

    def get_stdout(self, clear=True): 
     ret = self.unbblocked_out 
     if clear: 
      self.unbblocked_out = "" 
     return ret 

    def has_stdout(self): 
     ret = self.get_stdout(False) 
     if ret == '': 
      return None 
     else: 
      return ret 

    def get_stderr(self, clear=True): 
     ret = self.unbblocked_err 
     if clear: 
      self.unbblocked_err = "" 
     return ret 

    def has_stderr(self): 
     ret = self.get_stderr(False) 
     if ret == '': 
      return None 
     else: 
      return ret 

    def __init__(self, subp): 
     '''This is a simple class that collects and aggregates the 
     output from a subprocess so that you can more reliably use 
     the class without having to block for subprocess.communicate.''' 
     self.p = subp 
     self.unbblocked_out = "" 
     self.unbblocked_err = "" 
     self.running = True 
     self.qo = Queue.Queue() 
     self.to = threading.Thread(name="out_read", 
            target=self.enqueue_output, 
            args=()) 
     self.to.daemon = True # thread dies with the program 
     self.to.start() 

     self.qe = Queue.Queue() 
     self.te = threading.Thread(name="err_read", 
            target=self.enqueue_err, 
            args=()) 
     self.te.daemon = True # thread dies with the program 
     self.te.start() 

     self.stdin_queue = Queue.Queue() 
     self.aggregator = threading.Thread(name="aggregate", 
              target=self.aggregate, 
              args=()) 
     self.aggregator.daemon = True # thread dies with the program 
     self.aggregator.start() 
     pass 
def write_stdin(p,c): 
    while p.poll() == None: 
     i = raw_input("send to process:") 
     if i is not None: 
      c.stdin_queue.put(i) 


p = subprocess.Popen("cmd.exe", shell=True, stdout=subprocess.PIPE, 
        stderr=subprocess.PIPE, stdin=subprocess.PIPE) 
c = Process_Communicator(p) 
stdin = threading.Thread(name="write_stdin", 
          target=write_stdin, 
          args=(p,c)) 
stdin.daemon = True # thread dies with the program 
stdin.start() 
while p.poll() == None: 
    if c.has_stdout(): 
     print c.get_stdout() 
    if c.has_stderr(): 
     print c.get_stderr() 

c.join() 
print "Exit"