Differential D4231 Diff 10403 plugins/extensions/pykrita/plugin/plugins/scripter/debugger_scripter/debugger.py

Changeset View

Standalone View

View Options

plugins/extensions/pykrita/plugin/plugins/scripter/debugger_scripter/debugger.py

1import bdb 1​import bdb
2import asyncio 2​import asyncio
3import inspect 3​import inspect
4import multiprocessing 4​import multiprocessing
5​import re
5 6​
6 7​
7class Debugger(bdb.Bdb): 8​class Debugger(bdb.Bdb):
8 9​
9 def __init__(self, scripter, cmd): 10​ def __init__(self, scripter, cmd):
10 bdb.Bdb.__init__(self) 11​ bdb.Bdb.__init__(self)
11 12​
12 self.quit = False 13​ self.quit = False
13 self.debugq = multiprocessing.Queue() 14​ self.debugq = multiprocessing.Queue()
14 self.scripter = scripter 15​ self.scripter = scripter
15 self.applicationq = multiprocessing.Queue() 16​ self.applicationq = multiprocessing.Queue()
16 17​
17 # Create the debug process 18​ # Create the debug process
18 self.debugprocess = multiprocessing.Process(target=self.run, args=(cmd,)) 19​ self.debugprocess = multiprocessing.Process(target=self.run, args=(cmd,))
19 self.application_data = {} 20​ self.application_data = {}
20 self.currentLine = 0 21​ self.currentLine = 0
21 # initialize parent 22​ # initialize parent
22 bdb.Bdb.reset(self) 23​ bdb.Bdb.reset(self)
23 24​
24 def user_call(self, frame, args): 25​ def user_call(self, frame, args):
25 name = frame.f_code.co_name or "<unknown>" 26​ name = frame.f_code.co_name or "<unknown>"
26 27​
27 def user_line(self, frame): 28​ def user_line(self, frame):
28 """Handler that executes with every line of code""" 29​ """Handler that executes with every line of code"""
29 self.setCurrentLine(frame.f_lineno) 30​ co = frame.f_code
30 self.applicationq.put({ "lineNumber": self.getCurrentLine()}) 31​ self.currentLine = frame.f_lineno
32​ self.applicationq.put({ "code": { "file": co.co_filename,
33​ "name": co.co_name,
34​ "lineNumber": str(frame.f_lineno)
35​ },
36​ "frame": { "firstLineNumber": co.co_firstlineno,
37​ "locals": self.format_data(frame.f_locals),
38​ "globals": self.format_data(frame.f_globals)
39​ },
40​ "trace": "line"
41​ })
31 42​
32 if self.quit: 43​ if self.quit:
33 return self.set_quit() 44​ return self.set_quit()
34 45​
35 if self.getCurrentLine()==0: 46​ if self.currentLine==0:
36 return 47​ return
37 else: 48​ else:
38 # Get a reference to the code object and source 49​ # Get a reference to the code object and source
39 co = frame.f_code ​
40 source = inspect.getsourcelines(co)[0] 50​ source = inspect.getsourcelines(co)[0]
41 51​
42 # Wait for a debug command 52​ # Wait for a debug command
43 cmd = self.debugq.get() 53​ cmd = self.debugq.get()
44 54​
45 if cmd == "step": 55​ if cmd == "step":
46 # If stepping through code, return this handler 56​ # If stepping through code, return this handler
47 return 57​ return
48 58​
49 if cmd == "stop": 59​ if cmd == "stop":
50 # If stopping execution, raise an exception 60​ # If stopping execution, raise an exception
51 return self.set_quit() 61​ return self.set_quit()
52 62​
53 def user_return(self, frame, value): 63​ def user_return(self, frame, value):
54 name = frame.f_code.co_name or "<unknown>" 64​ name = frame.f_code.co_name or "<unknown>"
55 if name == '<module>': 65​ if name == '<module>':
56 self.applicationq.put({ "quit": True}) 66​ self.applicationq.put({ "quit": True})
57 67​
58 def user_exception(self, frame, exception): 68​ def user_exception(self, frame, exception):
59 name = frame.f_code.co_name or "<unknown>" 69​ name = frame.f_code.co_name or "<unknown>"
60 70​
61 def getCurrentLine(self): 71​ def format_data(self, data):
62 return self.currentLine 72​ globals()['types'] = __import__('types')
73​
74​ exclude_keys = ['copyright', 'credits', 'False',
75​ 'True', 'None', 'Ellipsis', 'quit']
76​ exclude_valuetypes = [types.BuiltinFunctionType,
77​ types.BuiltinMethodType,
78​ types.ModuleType,
79​ types.FunctionType]
80​
81​ return [{k: v} for k, v in data.items() if not (k in exclude_keys or
82​ type(v) in exclude_valuetypes or
83​ re.search(r'^(__).*\1$', k))]
63 84​
64 def setCurrentLine(self, line): ​
65 self.currentLine = line ​
66 85​
67 async def display(self): 86​ async def display(self):
68 """Coroutine for updating the UI""" 87​ """Coroutine for updating the UI"""
69 88​
70 # Wait for the application queue to have an update to the GUI 89​ # Wait for the application queue to have an update to the GUI
71 while True: 90​ while True:
72 if self.applicationq.empty(): 91​ if self.applicationq.empty():
73 await asyncio.sleep(0.5) 92​ await asyncio.sleep(0.3)
74 else: 93​ else:
75 # The application queue has at least one item, let's act on every item that's in it 94​ # The application queue has at least one item, let's act on every item that's in it
76 while not self.applicationq.empty(): 95​ while not self.applicationq.empty():
77 # Get info to the GUI 96​ # Get info to the GUI
78 self.application_data = self.applicationq.get() 97​ self.application_data = self.applicationq.get()
79 self.scripter.uicontroller.repaintDebugArea() 98​ self.scripter.uicontroller.repaintDebugArea()
80 return 99​ return
81 100​
82 async def start(self): 101​ async def start(self):
83 await self.display() 102​ await self.display()
84 103​
85 async def step(self): 104​ async def step(self):
86 # Tell the debugger we want to step in 105​ # Tell the debugger we want to step in
87 self.debugq.put("step") 106​ self.debugq.put("step")
88 107​
89 await self.display() 108​ await self.display()
90 109​
91 async def stop(self): 110​ async def stop(self):
92 # Tell the debugger we're stopping execution 111​ # Tell the debugger we're stopping execution
93 self.debugq.put("stop") 112​ self.debugq.put("stop")
113​ self.applicationq.put({ "quit": True})
114​ await self.display()