中止对 python 的交互式控制台的评估
Abort evaluation of python's interactive console
为了好玩，我正在编写自己的 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。
尝试：
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.
我不认为你可以轻易地在 Python 中杀死一个线程。不过，你可以杀死一个 multiprocessing.Process。因此，您可以使用单独的进程在控制台中执行代码并通过 multiprocessing.Queue 与其通信。为此，我实现了一个 TerminalManager class，它可以在单独的进程中执行 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()
为了好玩，我正在编写自己的 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。
尝试：
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.
我不认为你可以轻易地在 Python 中杀死一个线程。不过，你可以杀死一个 multiprocessing.Process。因此，您可以使用单独的进程在控制台中执行代码并通过 multiprocessing.Queue 与其通信。为此，我实现了一个 TerminalManager class，它可以在单独的进程中执行 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()