1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
from __future__ import nested_scopes
def set_trace_in_qt():
import pydevd_tracing
from pydevd_comm import GetGlobalDebugger
debugger = GetGlobalDebugger()
if debugger is not None:
pydevd_tracing.SetTrace(debugger.trace_dispatch)
_patched_qt = False
def patch_qt():
'''
This method patches qt (PySide or PyQt4) so that we have hooks to set the tracing for QThread.
'''
# Avoid patching more than once
global _patched_qt
if _patched_qt:
return
_patched_qt = True
try:
from PySide import QtCore
except:
try:
from PyQt4 import QtCore
except:
return
_original_thread_init = QtCore.QThread.__init__
_original_runnable_init = QtCore.QRunnable.__init__
class FuncWrapper:
def __init__(self, original):
self._original = original
def __call__(self, *args, **kwargs):
set_trace_in_qt()
return self._original(*args, **kwargs)
class StartedSignalWrapper: # Wrapper for the QThread.started signal
def __init__(self, thread, original_started):
self.thread = thread
self.original_started = original_started
def connect(self, func, *args, **kwargs):
return self.original_started.connect(FuncWrapper(func), *args, **kwargs)
def disconnect(self, *args, **kwargs):
return self.original_started.disconnect(*args, **kwargs)
def emit(self, *args, **kwargs):
return self.original_started.emit(*args, **kwargs)
class ThreadWrapper(QtCore.QThread): # Wrapper for QThread
def __init__(self, *args, **kwargs):
_original_thread_init(self)
self._original_run = self.run
self.run = self._new_run
self._original_started = self.started
self.started = StartedSignalWrapper(self, self.started)
def _new_run(self):
set_trace_in_qt()
return self._original_run()
class RunnableWrapper(QtCore.QRunnable): # Wrapper for QRunnable
def __init__(self, *args, **kwargs):
_original_runnable_init(self)
self._original_run = self.run
self.run = self._new_run
def _new_run(self):
set_trace_in_qt()
return self._original_run()
QtCore.QThread = ThreadWrapper
QtCore.QRunnable = RunnableWrapper
|
