Pythonのインタラクティブコンソールの評価を中止しました

Question

私自身のPythonコードエディタと端末を作成して、それを既存のプログラムに実装してスクライブ可能性を追加しています。

今は、実行中のコードの評価を停止する方法がわからないという問題を発見しました。どうすればそれができますか？

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経由でコードと通信できます。これを行うために、PythonTerminal.run_codeを別のプロセスで実行して終了できるTerminalManagerクラスを実装しました。以下の変更されたコードを参照してください。主な理由は、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 

これらの2つの方法の使用は、以下のとおりです。

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