中止评估python的交互式控制台

Question

我正在编写我自己的python代码编辑器和终端以获得乐趣，并在现有程序中实现它以添加scribtability。

现在我发现，我不知道该怎么一旦停止运行代码的评价问题。这怎么可能完成？

这是我实现：

import code 
import contextlib 
import sys 
from io import StringIO 
import copy 


@contextlib.contextmanager 
def capture(): 
    oldout,olderr = sys.stdout, sys.stderr 
    try: 
     out=[StringIO(), StringIO()] 
     sys.stdout,sys.stderr = out 
     yield out 
    finally: 
     sys.stdout,sys.stderr = oldout, olderr 
     out[0] = out[0].getvalue() 
     out[1] = out[1].getvalue() 


class PythonTerminal(code.InteractiveConsole): 

    def __init__(self, shared_vars): 
     self.shared_vars_start = copy.deepcopy(shared_vars) 
     self.shared_vars = shared_vars 
     super().__init__(shared_vars) 
     self.out_history = [] 

    def run_code(self,code_string): 
     with capture() as out: 
      self.runcode(code_string) 

     self.out_history.append(out) 
     return out 

    def restart_interpreter(self): 
     self.__init__(self.shared_vars_start) 

    def stop(self): 
     raise NotImplementedError 

if __name__ == '__main__': 
    a = range(10) 
    PyTerm = PythonTerminal({'Betrag': a}) 
    test_code = """ 
for i in range(10000): 
    for j in range(1000): 
     temp = i*j 
print('Finished'+str(i)) 
""" 
    print('Starting') 
    t = threading.Thread(target=PyTerm.run_code,args=(test_code,)) 
    t.start() 

    PyTerm.stop() 
    t.join() 
    print(PyTerm.out_history[-1]) # This line should be executed immediately and contain an InterruptError 

的目标是，评估停止，但该解释还活着，所以像CTRL + C。

Answer 1

我不认为你可以轻易的杀死Python中的线程。不过，你可以杀死一个multiprocessing.Process。因此，您可以使用单独的过程在控制台中执行代码，并通过multiprocessing.Queue与它进行通信。为此，我实现了一个TerminalManager类，可以在单独的进程中执行PythonTerminal.run_code并终止它。请参阅下面的修改代码。主要的缺点是，InteractiveConsole的本地人不会在两次通话之间持续存在。我已经添加了一个黑客（这可能是可怕的），将当地人存储到一个搁置文件。想到最快的东西。

import code 
import contextlib 
import sys 
from io import StringIO 
import copy 
import threading 
import multiprocessing 
import json 
import shelve 

class QueueIO: 
    """Uses a multiprocessing.Queue object o capture stdout and stderr""" 
    def __init__(self, q=None): 

     self.q = multiprocessing.Queue() if q is None else q 

    def write(self, value): 
     self.q.put(value) 

    def writelines(self, values): 
     self.q.put("\n".join(str(v) for v in values)) 

    def read(self): 
     return self.q.get() 

    def readlines(self): 
     result = "" 
     while not self.q.empty(): 
      result += self.q.get() + "\n" 


@contextlib.contextmanager 
def capture2(q: multiprocessing.Queue): 
    oldout,olderr = sys.stdout, sys.stderr 
    try: 
     qio = QueueIO(q) 
     out=[qio, qio] 
     sys.stdout,sys.stderr = out 
     yield out 
    finally: 
     sys.stdout,sys.stderr = oldout, olderr 


class PythonTerminal(code.InteractiveConsole): 

    def __init__(self, shared_vars): 
     self.shared_vars_start = copy.deepcopy(shared_vars) 
     self.shared_vars = shared_vars 
     super().__init__(shared_vars) 
     self.out_history = [] 

    def run_code(self,code_string, q): 
     # retrieve locals 
     d = shelve.open(r'd:\temp\shelve.pydb') 
     for k, v in d.items(): 
      self.locals[k] = v 

     # execute code 
     with capture2(q) as out: 
      self.runcode(code_string)    

     # store locals 
     for k, v in self.locals.items(): 
      try: 
       if k != '__builtins__': 
        d[k] = v 
      except TypeError: 
       pass 
     d.close() 


    def restart_interpreter(self): 
     self.__init__(self.shared_vars_start) 


class TerminalManager(): 

    def __init__(self, terminal): 
     self.terminal = terminal 
     self.process = None 
     self.q = multiprocessing.Queue()   

    def run_code(self, test_code): 
     self.process = multiprocessing.Process(
      target=self.terminal.run_code,args=(test_code, self.q)) 
     self.process.start() 

    def stop(self): 
     self.process.terminate() 
     self.q.put(repr(Exception('User interupted execution.'))) 

    def wait(self): 
     if self.process.is_alive: 
      self.process.join() 
     while not self.q.empty(): 
      print(self.q.get())  

if __name__ == '__main__': 
    import time 
    a = range(10) 
    PyTerm = PythonTerminal({'Betrag': a}) 
    test_code = """ 
import time 
a = 'hello' 
for i in range(10): 
    time.sleep(0.2) 
    print(i) 
print('Finished') 
""" 
    mgr = TerminalManager(PyTerm) 
    print('Starting') 
    mgr.run_code(test_code)  
    time.sleep(1) 
    mgr.stop() 
    mgr.wait() 

    test_code = """ 
import time 
_l = locals() 

print('a = {}'.format(a)) 
for i in range(10): 
    time.sleep(0.1) 
    print(i) 
print('Finished') 
""" 
    print('Starting again') 
    mgr.run_code(test_code)   

    mgr.wait() 

Answer 2

尝试：

def stop(self): 
    self.resetbuffer()#abort currently executing code by wiping the input buffer 
    self.push("exit()")#trigger an exit from the interpreter 

这两种方法的用法如下：

| push(self, line) 
|  Push a line to the interpreter. 
|  
|  The line should not have a trailing newline; it may have 
|  internal newlines. The line is appended to a buffer and the 
|  interpreter's runsource() method is called with the 
|  concatenated contents of the buffer as source. If this 
|  indicates that the command was executed or invalid, the buffer 
|  is reset; otherwise, the command is incomplete, and the buffer 
|  is left as it was after the line was appended. The return 
|  value is 1 if more input is required, 0 if the line was dealt 
|  with in some way (this is the same as runsource()). 

| resetbuffer(self) 
|  Reset the input buffer.