中止評估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.