131 lines
3.2 KiB
Python
131 lines
3.2 KiB
Python
# Forked from https://raw.githubusercontent.com/romanvm/python-web-pdb
|
||
from __future__ import absolute_import
|
||
from __future__ import unicode_literals
|
||
|
||
import sys
|
||
import weakref
|
||
from threading import Event
|
||
from threading import RLock
|
||
|
||
try:
|
||
import queue
|
||
except ImportError:
|
||
import Queue as queue
|
||
|
||
|
||
class ThreadSafeObject(object):
|
||
"""
|
||
An object for data exchange between threads
|
||
"""
|
||
|
||
def __init__(self, contents=None):
|
||
self._lock = RLock()
|
||
self._contents = contents
|
||
|
||
@property
|
||
def contents(self):
|
||
with self._lock:
|
||
return self._contents
|
||
|
||
@contents.setter
|
||
def contents(self, value):
|
||
with self._lock:
|
||
self._contents = value
|
||
|
||
|
||
class Console(object):
|
||
input_queue = queue.Queue()
|
||
|
||
def __init__(self, debugger):
|
||
self.pdb_not_busy = Event()
|
||
self.closed = False
|
||
self._debugger = weakref.proxy(debugger)
|
||
self._console_history = ThreadSafeObject("")
|
||
self._frame_data = ThreadSafeObject({})
|
||
self._output_ok = Event()
|
||
self.exception = None
|
||
|
||
|
||
@property
|
||
def seekable(self):
|
||
return False
|
||
|
||
@property
|
||
def writable(self):
|
||
return True
|
||
|
||
@property
|
||
def encoding(self):
|
||
return "utf-8"
|
||
|
||
def close(self):
|
||
self.closed = True
|
||
self.pdb_not_busy.set()
|
||
|
||
# console每次读取从输入队列读一行，每读到一行就往commands中写一行，并且往console写，获取响应栈帧
|
||
def readline(self):
|
||
self.pdb_not_busy.set()
|
||
command = self.input_queue.get()
|
||
self.pdb_not_busy.clear()
|
||
|
||
if not command:
|
||
command = "\n"
|
||
|
||
with open("commands.txt", "a+") as file:
|
||
file.write(command)
|
||
self.writeline(command)
|
||
return command
|
||
|
||
read = readline
|
||
|
||
# 输出到console
|
||
def writeline(self, data):
|
||
# with self._output_lock:
|
||
if isinstance(data, bytes):
|
||
data = data.decode("utf-8")
|
||
|
||
self._console_history.contents += data
|
||
try:
|
||
frame_data = self._debugger.get_current_frame_data()
|
||
except (IOError, AttributeError):
|
||
f = open("user_code/user.out")
|
||
out_content = f.read()
|
||
f.close()
|
||
frame_data = {
|
||
"dirname": "",
|
||
"filename": "",
|
||
"file_listing": "No data available",
|
||
"current_line": -1,
|
||
"breakpoints": {},
|
||
"globals": {},
|
||
"locals": {},
|
||
"expressions": {},
|
||
"exception": sys.exc_info(),
|
||
"out": out_content,
|
||
"is_over": True
|
||
}
|
||
|
||
frame_data["console_history"] = self._console_history.contents
|
||
self._frame_data.contents = frame_data
|
||
if "(Pdb)" in data:
|
||
self._output_ok.set()
|
||
|
||
write = writeline
|
||
|
||
def flush(self):
|
||
pass
|
||
|
||
def send_pdb_command(self, command):
|
||
self.pdb_not_busy.wait()
|
||
self.input_queue.put(command)
|
||
self._output_ok.clear()
|
||
# print("设置了command: " + command)
|
||
|
||
def get_frame_data(self):
|
||
# print("f 尝试获取琐")
|
||
self._output_ok.wait(3)
|
||
# print("f 获取到琐")
|
||
c = self._frame_data.contents
|
||
return c
|
||
|