245 lines
8.0 KiB
Python
245 lines
8.0 KiB
Python
# Forked from https://raw.githubusercontent.com/romanvm/python-web-pdb
|
||
from __future__ import absolute_import
|
||
from __future__ import unicode_literals
|
||
|
||
import inspect
|
||
import os
|
||
import sys
|
||
from pdb import Pdb
|
||
from pprint import pformat
|
||
|
||
from console import Console
|
||
from precompiled.__serializer__ import __Serializer__
|
||
|
||
|
||
class PdbWrapper(Pdb):
|
||
active_instance = None
|
||
null = object()
|
||
|
||
def __init__(self, stdout=None, stderr=None, stdin=None):
|
||
# 初始化console对象及PDB对象，实际做的事情是标准输入从input queue读取，标准输出写到pdb的console，备份标准流
|
||
self.console = Console(self)
|
||
self.expressions = []
|
||
Pdb.__init__(self, stdin=self.console, stdout=self.console)
|
||
|
||
self._backup = []
|
||
|
||
if stdout is not None:
|
||
self._backup.append(("stdout", sys.stdout))
|
||
setattr(sys, "stdout", stdout)
|
||
|
||
if stderr is not None:
|
||
self._backup.append(("stderr", sys.stderr))
|
||
setattr(sys, "stderr", stderr)
|
||
|
||
if stdin is not None:
|
||
self._backup.append(("stdin", sys.stdin))
|
||
setattr(sys, "stdin", stdin)
|
||
|
||
PdbWrapper.active_instance = self
|
||
|
||
def do_quit(self, arg):
|
||
"""
|
||
quit || exit || q
|
||
Stop and quit the current debugging session
|
||
"""
|
||
for name, fh in self._backup:
|
||
setattr(sys, name, fh)
|
||
self.console.writeline("*** Aborting program ***\n")
|
||
self.console.flush()
|
||
self.console.close()
|
||
PdbWrapper.active_instance = None
|
||
return Pdb.do_quit(self, arg)
|
||
|
||
do_q = do_exit = do_quit
|
||
|
||
def do_inspect(self, arg):
|
||
"""
|
||
i(nspect) object
|
||
Inspect an object
|
||
"""
|
||
if arg in self.curframe.f_locals:
|
||
obj = self.curframe.f_locals[arg]
|
||
elif arg in self.curframe.f_globals:
|
||
obj = self.curframe.f_globals[arg]
|
||
else:
|
||
obj = PdbWrapper.null
|
||
if obj is not PdbWrapper.null:
|
||
self.console.writeline("{0} = {1}:\n".format(arg, type(obj)))
|
||
for name, value in inspect.getmembers(obj):
|
||
if not (name.startswith("__") and (name.endswith("__"))):
|
||
self.console.writeline(
|
||
" {0}: {1}\n".format(
|
||
name, self._get_repr(value, pretty=True, indent=8)
|
||
)
|
||
)
|
||
else:
|
||
self.console.writeline(
|
||
'NameError: name "{0}" is not defined\n'.format(arg)
|
||
)
|
||
self.console.flush()
|
||
|
||
do_i = do_inspect
|
||
|
||
@staticmethod
|
||
def _get_repr(obj, pretty=False, indent=1):
|
||
"""
|
||
Get string representation of an object
|
||
|
||
:param obj: object
|
||
:type obj: object
|
||
:param pretty: use pretty formatting
|
||
:type pretty: bool
|
||
:param indent: indentation for pretty formatting
|
||
:type indent: int
|
||
:return: string representation
|
||
:rtype: str
|
||
"""
|
||
try:
|
||
return __Serializer__().serialize(obj)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
if pretty:
|
||
repr_value = pformat(obj, indent)
|
||
else:
|
||
repr_value = repr(obj)
|
||
except Exception:
|
||
return obj.__class__.__name__
|
||
|
||
if sys.version_info[0] == 2:
|
||
# Try to convert Unicode string to human-readable form
|
||
try:
|
||
repr_value = repr_value.decode("raw_unicode_escape")
|
||
except UnicodeError:
|
||
repr_value = repr_value.decode("utf-8", "replace")
|
||
return repr_value
|
||
|
||
def set_continue(self):
|
||
# We do not detach the debugger
|
||
# for correct multiple set_trace() and post_mortem() calls.
|
||
self._set_stopinfo(self.botframe, None, -1)
|
||
|
||
def dispatch_return(self, frame, arg):
|
||
# The parent's method needs to be called first.
|
||
ret = Pdb.dispatch_return(self, frame, arg)
|
||
if frame.f_back is None:
|
||
self.console.writeline("*** Thread finished ***\n")
|
||
if not self.console.closed:
|
||
self.console.flush()
|
||
self.console.close()
|
||
return ret
|
||
|
||
def get_current_frame_data(self):
|
||
"""
|
||
Get all data about the current execution frame
|
||
|
||
:return: current frame data
|
||
:rtype: dict
|
||
:raises AttributeError: if the debugger does hold any execution frame.
|
||
:raises IOError: if source code for the current execution frame is not
|
||
accessible.
|
||
"""
|
||
filename = self.curframe.f_code.co_filename
|
||
lines, start_line = inspect.findsource(self.curframe)
|
||
if sys.version_info[0] == 2:
|
||
lines = [line.decode("utf-8") for line in lines]
|
||
return {
|
||
"dirname": os.path.dirname(os.path.abspath(filename)) + os.path.sep,
|
||
"filename": os.path.basename(filename),
|
||
"file_listing": "".join(lines),
|
||
"current_line": self.curframe.f_lineno,
|
||
"breakpoints": self.get_file_breaks("user_code/prog_joined.py"),
|
||
"globals": self.get_globals(),
|
||
"locals": self.get_locals(),
|
||
"expressions": self._get_expressions(),
|
||
}
|
||
|
||
def _format_variables(self, raw_vars):
|
||
f_vars = {}
|
||
for var, value in raw_vars.items():
|
||
if not (var.startswith("__") and var.endswith("__")):
|
||
repr_value = self._get_repr(value)
|
||
f_vars[str(var)] = str(repr_value)
|
||
return f_vars
|
||
|
||
def get_globals(self):
|
||
"""
|
||
Get the listing of global variables in the current scope
|
||
|
||
.. note:: special variables that start and end with
|
||
double underscores ``__`` are not included.
|
||
|
||
:return: a listing of ``var = value`` pairs sorted alphabetically
|
||
:rtype: unicode
|
||
"""
|
||
return self._format_variables(self.curframe.f_globals)
|
||
|
||
def get_locals(self):
|
||
"""
|
||
Get the listing of local variables in the current scope
|
||
|
||
.. note:: special variables that start and end with
|
||
double underscores ``__`` are not included.
|
||
For module scope globals and locals listings are the same.
|
||
|
||
:return: a listing of ``var = value`` pairs sorted alphabetically
|
||
:rtype: unicode
|
||
"""
|
||
return self._format_variables(self.curframe.f_locals)
|
||
|
||
def remove_trace(self, frame=None):
|
||
"""
|
||
Detach the debugger from the execution stack
|
||
|
||
:param frame: the lowest frame to detach the debugger from.
|
||
:type frame: types.FrameType
|
||
"""
|
||
sys.settrace(None)
|
||
if frame is None:
|
||
frame = self.curframe
|
||
while frame and frame is not self.botframe:
|
||
del frame.f_trace
|
||
frame = frame.f_back
|
||
|
||
def debug(self):
|
||
self.set_trace(sys._getframe().f_back)
|
||
|
||
def get_console(self):
|
||
return self.console
|
||
|
||
def add_expression(self, expression):
|
||
if expression not in self._get_expressions().keys():
|
||
self.expressions.append(expression)
|
||
|
||
def remove_expression(self, expression):
|
||
try:
|
||
self.expressions.remove(expression)
|
||
except ValueError:
|
||
pass
|
||
|
||
def _eval(self, src, *args, **kwargs):
|
||
return eval(compile(src, "<stdin>", "eval"), *args, **kwargs)
|
||
|
||
def _get_expressions(self):
|
||
expression_dict = {}
|
||
for expression in self.expressions:
|
||
try:
|
||
value = self._eval(
|
||
expression, self.curframe.f_globals, self.curframe.f_locals
|
||
)
|
||
expression_dict[expression] = self._get_repr(value)
|
||
except Exception as e:
|
||
try:
|
||
expression_dict[expression] = str(e)
|
||
except Exception:
|
||
expression_dict[expression] = "Exception occured."
|
||
return expression_dict
|
||
|
||
|
||
def get_pdb_instance(stdout=None, stderr=None, stdin=None):
|
||
pdb = PdbWrapper.active_instance
|
||
return pdb if pdb else PdbWrapper(stdout=stdout, stderr=stderr, stdin=stdin)
|
||
|