142 lines
3.8 KiB
Python
142 lines
3.8 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
|
||
from pathlib import Path
|
||
|
||
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):
|
||
frame_data = {
|
||
"dirname": "",
|
||
"filename": "",
|
||
"file_listing": "No data available",
|
||
"current_line": -1,
|
||
"breakpoints": [],
|
||
"globals": {},
|
||
"locals": {},
|
||
"expressions": {},
|
||
"exception": sys.exc_info(),
|
||
}
|
||
|
||
frame_data["console_history"] = self._console_history.contents
|
||
frame_data["stdout"] = Path("user_code/stdout.txt").read_text().replace("""
|
||
PYDEV DEBUGGER WARNING:
|
||
sys.settrace() should not be used when the debugger is being used.
|
||
This may cause the debugger to stop working correctly.
|
||
If this is needed, please check:
|
||
http://pydev.blogspot.com/2007/06/why-cant-pydev-debugger-work-with.html
|
||
to see how to restore the debug tracing back correctly.
|
||
Call Location:
|
||
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/bdb.py", line 334, in set_trace
|
||
sys.settrace(self.trace_dispatch)
|
||
|
||
""", "")
|
||
self._frame_data.contents = frame_data
|
||
# print("frame_data: " + str(frame_data))
|
||
# print("data: " + data)
|
||
if "(Pdb)" in data:
|
||
self._output_ok.set()
|
||
# print("释放OK琐")
|
||
|
||
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()
|
||
# print("f 获取到琐")
|
||
c = self._frame_data.contents
|
||
return c
|
||
|