中止对 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

试试：

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.

Answer 2

我认为你不能轻易地杀死 Python 中的线程。不过，您可以杀死 multiprocessing.Process。因此，您可以使用单独的进程在控制台中执行代码并通过multiprocessing.Queue 与其通信。为此，我实现了一个 TerminalManager 类，它可以在单独的进程中执行 PythonTerminal.run_code 并将其杀死。请参阅下面的修改后的代码。一个主要的缺点是 InteractiveConsole 的本地变量不会在调用之间持续存在。我添加了一个 hack（这可能很糟糕），将本地人存储到一个搁置文件中。想到的最快的事情。

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()